From c841cfc8fc801df78469b616ace4816daadb3615 Mon Sep 17 00:00:00 2001
From: None <code@bormenusa.com>
Date: Mon, 23 Feb 2026 00:34:28 -0600
Subject: [PATCH] Fix #861: Prevent M3U endless downloading caused by Redis
 lock expiry
MIME-Version: 1.0
Content-Type: text/plain; charset=UTF-8
Content-Transfer-Encoding: 8bit

Large M3U downloads that exceed the 300s Redis lock TTL caused Celery Beat
to re-queue duplicate tasks, creating overlapping downloads that never complete.

Files changed:

- core/utils.py: Add TaskLockRenewer class — a daemon thread that periodically renews Redis lock TTL (every 120s) to prevent expiry during long-running tasks.

- apps/m3u/tasks.py: Apply TaskLockRenewer to refresh_single_m3u_account and refresh_m3u_groups; add HTTP timeout (30s connect, 60s read) to M3U download (the only download path missing one); stream M3U download to disk instead of accumulating in memory; add Celery task time limits (1 hour hard limit).

- apps/epg/tasks.py: Apply TaskLockRenewer to refresh_epg_data and parse_programs_for_tvg_id; add Celery task time limits (30 min / 1 hour).

Verified with a throttled test server: locks renewed at T+120s, T+240s, T+360s; 50,000 streams processed with no duplicate tasks.
---
 apps/epg/tasks.py |  26 +++-
 apps/m3u/tasks.py | 301 ++++++++++++++++++++++++++--------------------
 core/utils.py     |  69 +++++++++++
 3 files changed, 264 insertions(+), 132 deletions(-)

diff --git a/apps/epg/tasks.py b/apps/epg/tasks.py
index 97552171..9dc597d3 100644
--- a/apps/epg/tasks.py
+++ b/apps/epg/tasks.py
@@ -24,7 +24,7 @@ from asgiref.sync import async_to_sync
 from channels.layers import get_channel_layer
 
 from .models import EPGSource, EPGData, ProgramData
-from core.utils import acquire_task_lock, release_task_lock, send_websocket_update, cleanup_memory, log_system_event
+from core.utils import acquire_task_lock, release_task_lock, TaskLockRenewer, send_websocket_update, cleanup_memory, log_system_event
 
 logger = logging.getLogger(__name__)
 
@@ -146,12 +146,15 @@ def refresh_all_epg_data():
     return "EPG data refreshed."
 
 
-@shared_task
+@shared_task(time_limit=1800, soft_time_limit=1700)
 def refresh_epg_data(source_id):
     if not acquire_task_lock('refresh_epg_data', source_id):
         logger.debug(f"EPG refresh for {source_id} already running")
         return
 
+    lock_renewer = TaskLockRenewer('refresh_epg_data', source_id)
+    lock_renewer.start()
+
     source = None
     try:
         # Try to get the EPG source
@@ -168,6 +171,7 @@ def refresh_epg_data(source_id):
                 logger.info(f"No orphaned task found for EPG source {source_id}")
 
             # Release the lock and exit
+            lock_renewer.stop()
             release_task_lock('refresh_epg_data', source_id)
             # Force garbage collection before exit
             gc.collect()
@@ -176,6 +180,7 @@ def refresh_epg_data(source_id):
         # The source exists but is not active, just skip processing
         if not source.is_active:
             logger.info(f"EPG source {source_id} is not active. Skipping.")
+            lock_renewer.stop()
             release_task_lock('refresh_epg_data', source_id)
             # Force garbage collection before exit
             gc.collect()
@@ -184,6 +189,7 @@ def refresh_epg_data(source_id):
         # Skip refresh for dummy EPG sources - they don't need refreshing
         if source.source_type == 'dummy':
             logger.info(f"Skipping refresh for dummy EPG source {source.name} (ID: {source_id})")
+            lock_renewer.stop()
             release_task_lock('refresh_epg_data', source_id)
             gc.collect()
             return
@@ -194,6 +200,7 @@ def refresh_epg_data(source_id):
             fetch_success = fetch_xmltv(source)
             if not fetch_success:
                 logger.error(f"Failed to fetch XMLTV for source {source.name}")
+                lock_renewer.stop()
                 release_task_lock('refresh_epg_data', source_id)
                 # Force garbage collection before exit
                 gc.collect()
@@ -202,6 +209,7 @@ def refresh_epg_data(source_id):
             parse_channels_success = parse_channels_only(source)
             if not parse_channels_success:
                 logger.error(f"Failed to parse channels for source {source.name}")
+                lock_renewer.stop()
                 release_task_lock('refresh_epg_data', source_id)
                 # Force garbage collection before exit
                 gc.collect()
@@ -234,6 +242,7 @@ def refresh_epg_data(source_id):
         source = None
         # Force garbage collection before releasing the lock
         gc.collect()
+        lock_renewer.stop()
         release_task_lock('refresh_epg_data', source_id)
 
 
@@ -1126,12 +1135,15 @@ def parse_channels_only(source):
 
 
 
-@shared_task
+@shared_task(time_limit=3600, soft_time_limit=3500)
 def parse_programs_for_tvg_id(epg_id):
     if not acquire_task_lock('parse_epg_programs', epg_id):
         logger.info(f"Program parse for {epg_id} already in progress, skipping duplicate task")
         return "Task already running"
 
+    lock_renewer = TaskLockRenewer('parse_epg_programs', epg_id)
+    lock_renewer.start()
+
     source_file = None
     program_parser = None
     programs_to_create = []
@@ -1161,11 +1173,13 @@ def parse_programs_for_tvg_id(epg_id):
         # Skip program parsing for dummy EPG sources - they don't have program data files
         if epg_source.source_type == 'dummy':
             logger.info(f"Skipping program parsing for dummy EPG source {epg_source.name} (ID: {epg_id})")
+            lock_renewer.stop()
             release_task_lock('parse_epg_programs', epg_id)
             return
 
         if not Channel.objects.filter(epg_data=epg).exists():
             logger.info(f"No channels matched to EPG {epg.tvg_id}")
+            lock_renewer.stop()
             release_task_lock('parse_epg_programs', epg_id)
             return
 
@@ -1207,6 +1221,7 @@ def parse_programs_for_tvg_id(epg_id):
                     epg_source.last_message = f"Failed to download EPG data, cannot parse programs"
                     epg_source.save(update_fields=['status', 'last_message'])
                     send_epg_update(epg_source.id, "parsing_programs", 100, status="error", error="Failed to download EPG file")
+                    lock_renewer.stop()
                     release_task_lock('parse_epg_programs', epg_id)
                     return
 
@@ -1217,6 +1232,7 @@ def parse_programs_for_tvg_id(epg_id):
                     epg_source.last_message = f"Failed to download EPG data, file missing after download"
                     epg_source.save(update_fields=['status', 'last_message'])
                     send_epg_update(epg_source.id, "parsing_programs", 100, status="error", error="File not found after download")
+                    lock_renewer.stop()
                     release_task_lock('parse_epg_programs', epg_id)
                     return
 
@@ -1232,6 +1248,7 @@ def parse_programs_for_tvg_id(epg_id):
                 epg_source.last_message = f"No URL provided, cannot fetch EPG data"
                 epg_source.save(update_fields=['status', 'last_message'])
                 send_epg_update(epg_source.id, "parsing_programs", 100, status="error", error="No URL provided")
+                lock_renewer.stop()
                 release_task_lock('parse_epg_programs', epg_id)
                 return
 
@@ -1379,7 +1396,7 @@ def parse_programs_for_tvg_id(epg_id):
         epg_source = None
         # Add comprehensive cleanup before releasing lock
         cleanup_memory(log_usage=should_log_memory, force_collection=True)
-         # Memory tracking after processing
+        # Memory tracking after processing
         if process:
             try:
                 mem_after = process.memory_info().rss / 1024 / 1024
@@ -1389,6 +1406,7 @@ def parse_programs_for_tvg_id(epg_id):
             process = None
         epg = None
         programs_processed = None
+        lock_renewer.stop()
         release_task_lock('parse_epg_programs', epg_id)
 
 
diff --git a/apps/m3u/tasks.py b/apps/m3u/tasks.py
index dc178317..0f2bdf66 100644
--- a/apps/m3u/tasks.py
+++ b/apps/m3u/tasks.py
@@ -23,6 +23,7 @@ from core.utils import (
     RedisClient,
     acquire_task_lock,
     release_task_lock,
+    TaskLockRenewer,
     natural_sort_key,
     log_system_event,
 )
@@ -66,7 +67,8 @@ def fetch_m3u_lines(account, use_cache=False):
                 account.save(update_fields=["status", "last_message"])
 
                 response = requests.get(
-                    account.server_url, headers=headers, stream=True
+                    account.server_url, headers=headers, stream=True,
+                    timeout=(30, 60),  # 30s connect, 60s read between chunks
                 )
 
                 # Log the actual response details for debugging
@@ -126,119 +128,60 @@ def fetch_m3u_lines(account, use_cache=False):
                 start_time = time.time()
                 last_update_time = start_time
                 progress = 0
-                temp_content = b""  # Store content temporarily to validate before saving
                 has_content = False
 
-                # First, let's collect the content and validate it
-                send_m3u_update(account.id, "downloading", 0)
-                for chunk in response.iter_content(chunk_size=8192):
-                    if chunk:
-                        temp_content += chunk
-                        has_content = True
+                # Stream directly to a temp file to avoid holding the entire
+                # M3U in memory (large files can be 100MB+, which would use
+                # ~3x that in RAM in certain approaches).
+                temp_path = file_path + ".tmp"
+                try:
+                    send_m3u_update(account.id, "downloading", 0)
+                    with open(temp_path, "wb") as tmp_file:
+                        for chunk in response.iter_content(chunk_size=8192):
+                            if chunk:
+                                tmp_file.write(chunk)
+                                has_content = True
 
-                        downloaded += len(chunk)
-                        elapsed_time = time.time() - start_time
+                                downloaded += len(chunk)
+                                elapsed_time = time.time() - start_time
 
-                        # Calculate download speed in KB/s
-                        speed = downloaded / elapsed_time / 1024  # in KB/s
+                                # Calculate download speed in KB/s
+                                speed = downloaded / elapsed_time / 1024  # in KB/s
 
-                        # Calculate progress percentage
-                        if total_size and total_size > 0:
-                            progress = (downloaded / total_size) * 100
+                                # Calculate progress percentage
+                                if total_size and total_size > 0:
+                                    progress = (downloaded / total_size) * 100
 
-                        # Time remaining (in seconds)
-                        time_remaining = (
-                            (total_size - downloaded) / (speed * 1024)
-                            if speed > 0
-                            else 0
-                        )
-
-                        current_time = time.time()
-                        if current_time - last_update_time >= 0.5:
-                            last_update_time = current_time
-                            if progress > 0:
-                                # Update the account's last_message with detailed progress info
-                                progress_msg = f"Downloading: {progress:.1f}% - {speed:.1f} KB/s - {time_remaining:.1f}s remaining"
-                                account.last_message = progress_msg
-                                account.save(update_fields=["last_message"])
-
-                                send_m3u_update(
-                                    account.id,
-                                    "downloading",
-                                    progress,
-                                    speed=speed,
-                                    elapsed_time=elapsed_time,
-                                    time_remaining=time_remaining,
-                                    message=progress_msg,
+                                # Time remaining (in seconds)
+                                time_remaining = (
+                                    (total_size - downloaded) / (speed * 1024)
+                                    if speed > 0
+                                    else 0
                                 )
 
-                # Check if we actually received any content
-                logger.info(f"Download completed. Has content: {has_content}, Content length: {len(temp_content)} bytes")
-                if not has_content or len(temp_content) == 0:
-                    error_msg = f"Server responded successfully (HTTP {response.status_code}) but provided empty M3U file from URL: {account.server_url}"
-                    logger.error(error_msg)
-                    account.status = M3UAccount.Status.ERROR
-                    account.last_message = error_msg
-                    account.save(update_fields=["status", "last_message"])
-                    send_m3u_update(
-                        account.id,
-                        "downloading",
-                        100,
-                        status="error",
-                        error=error_msg,
-                    )
-                    return [], False
+                                current_time = time.time()
+                                if current_time - last_update_time >= 0.5:
+                                    last_update_time = current_time
+                                    if progress > 0:
+                                        # Update the account's last_message with detailed progress info
+                                        progress_msg = f"Downloading: {progress:.1f}% - {speed:.1f} KB/s - {time_remaining:.1f}s remaining"
+                                        account.last_message = progress_msg
+                                        account.save(update_fields=["last_message"])
 
-                # Basic validation: check if content looks like an M3U file
-                try:
-                    content_str = temp_content.decode('utf-8', errors='ignore')
-                    content_lines = content_str.strip().split('\n')
+                                        send_m3u_update(
+                                            account.id,
+                                            "downloading",
+                                            progress,
+                                            speed=speed,
+                                            elapsed_time=elapsed_time,
+                                            time_remaining=time_remaining,
+                                            message=progress_msg,
+                                        )
 
-                    # Log first few lines for debugging (be careful not to log too much)
-                    preview_lines = content_lines[:5]
-                    logger.info(f"Content preview (first 5 lines): {preview_lines}")
-                    logger.info(f"Total lines in content: {len(content_lines)}")
-
-                    # Check if it's a valid M3U file (should start with #EXTM3U or contain M3U-like content)
-                    is_valid_m3u = False
-
-                    # First, check if this looks like an error response disguised as 200 OK
-                    content_lower = content_str.lower()
-                    if any(error_indicator in content_lower for error_indicator in [
-                        '<html', '<!doctype html', 'error', 'not found', '404', '403', '500',
-                        'access denied', 'unauthorized', 'forbidden', 'invalid', 'expired'
-                    ]):
-                        logger.warning(f"Content appears to be an error response disguised as HTTP 200: {content_str[:200]!r}")
-                        # Continue with M3U validation, but this gives us a clue
-
-                    if content_lines and content_lines[0].strip().upper().startswith('#EXTM3U'):
-                        is_valid_m3u = True
-                        logger.info("Content validated as M3U: starts with #EXTM3U")
-                    elif any(line.strip().startswith('#EXTINF:') for line in content_lines):
-                        is_valid_m3u = True
-                        logger.info("Content validated as M3U: contains #EXTINF entries")
-                    elif any(line.strip().startswith('http') for line in content_lines):
-                        # Has HTTP URLs, might be a simple M3U without headers
-                        is_valid_m3u = True
-                        logger.info("Content validated as M3U: contains HTTP URLs")
-                    elif any(line.strip().startswith(('rtsp', 'rtp', 'udp')) for line in content_lines):
-                        # Has RTSP/RTP/UDP URLs, might be a simple M3U without headers
-                        is_valid_m3u = True
-                        logger.info("Content validated as M3U: contains RTSP/RTP/UDP URLs")
-
-                    if not is_valid_m3u:
-                        # Log what we actually received for debugging
-                        logger.error(f"Invalid M3U content received. First 200 characters: {content_str[:200]!r}")
-
-                        # Try to provide more specific error messages based on content
-                        if '<html' in content_lower or '<!doctype html' in content_lower:
-                            error_msg = f"Server returned HTML page instead of M3U file from URL: {account.server_url}. This usually indicates an error or authentication issue."
-                        elif 'error' in content_lower or 'not found' in content_lower:
-                            error_msg = f"Server returned an error message instead of M3U file from URL: {account.server_url}. Content: {content_str[:100]}"
-                        elif len(content_str.strip()) == 0:
-                            error_msg = f"Server returned completely empty response from URL: {account.server_url}"
-                        else:
-                            error_msg = f"Server provided invalid M3U content from URL: {account.server_url}. Content does not appear to be a valid M3U file."
+                    # Check if we actually received any content
+                    logger.info(f"Download completed. Has content: {has_content}, Content length: {downloaded} bytes")
+                    if not has_content or downloaded == 0:
+                        error_msg = f"Server responded successfully (HTTP {response.status_code}) but provided empty M3U file from URL: {account.server_url}"
                         logger.error(error_msg)
                         account.status = M3UAccount.Status.ERROR
                         account.last_message = error_msg
@@ -252,31 +195,113 @@ def fetch_m3u_lines(account, use_cache=False):
                         )
                         return [], False
 
-                except UnicodeDecodeError:
-                    logger.error(f"Non-text content received. First 200 bytes: {temp_content[:200]!r}")
-                    error_msg = f"Server provided non-text content from URL: {account.server_url}. Unable to process as M3U file."
-                    logger.error(error_msg)
-                    account.status = M3UAccount.Status.ERROR
-                    account.last_message = error_msg
-                    account.save(update_fields=["status", "last_message"])
-                    send_m3u_update(
-                        account.id,
-                        "downloading",
-                        100,
-                        status="error",
-                        error=error_msg,
-                    )
-                    return [], False
+                    # Validate the file by reading only the first portion from
+                    # disk — no need to load the entire file into memory just
+                    # to check the header.
+                    VALIDATION_READ_SIZE = 32768  # 32KB covers headers comfortably
+                    try:
+                        with open(temp_path, "rb") as vf:
+                            head_bytes = vf.read(VALIDATION_READ_SIZE)
+                        head_str = head_bytes.decode('utf-8', errors='ignore')
+                        head_lines = head_str.strip().split('\n')
 
-                # Content is valid, save it to file
-                with open(file_path, "wb") as file:
-                    file.write(temp_content)
+                        # Count total lines efficiently without loading full file
+                        with open(temp_path, "rb") as vf:
+                            total_lines = sum(1 for _ in vf)
 
-                # Final update with 100% progress
-                final_msg = f"Download complete. Size: {total_size/1024/1024:.2f} MB, Time: {time.time() - start_time:.1f}s"
-                account.last_message = final_msg
-                account.save(update_fields=["last_message"])
-                send_m3u_update(account.id, "downloading", 100, message=final_msg)
+                        # Log first few lines for debugging (be careful not to log too much)
+                        preview_lines = head_lines[:5]
+                        logger.info(f"Content preview (first 5 lines): {preview_lines}")
+                        logger.info(f"Total lines in content: {total_lines}")
+
+                        # Check if it's a valid M3U file (should start with #EXTM3U or contain M3U-like content)
+                        is_valid_m3u = False
+
+                        # First, check if this looks like an error response disguised as 200 OK
+                        head_lower = head_str.lower()
+                        if any(error_indicator in head_lower for error_indicator in [
+                            '<html', '<!doctype html', 'error', 'not found', '404', '403', '500',
+                            'access denied', 'unauthorized', 'forbidden', 'invalid', 'expired'
+                        ]):
+                            logger.warning(f"Content appears to be an error response disguised as HTTP 200: {head_str[:200]!r}")
+                            # Continue with M3U validation, but this gives us a clue
+
+                        if head_lines and head_lines[0].strip().upper().startswith('#EXTM3U'):
+                            is_valid_m3u = True
+                            logger.info("Content validated as M3U: starts with #EXTM3U")
+                        elif any(line.strip().startswith('#EXTINF:') for line in head_lines):
+                            is_valid_m3u = True
+                            logger.info("Content validated as M3U: contains #EXTINF entries")
+                        elif any(line.strip().startswith('http') for line in head_lines):
+                            # Has HTTP URLs, might be a simple M3U without headers
+                            is_valid_m3u = True
+                            logger.info("Content validated as M3U: contains HTTP URLs")
+                        elif any(line.strip().startswith(('rtsp', 'rtp', 'udp')) for line in head_lines):
+                            # Has RTSP/RTP/UDP URLs, might be a simple M3U without headers
+                            is_valid_m3u = True
+                            logger.info("Content validated as M3U: contains RTSP/RTP/UDP URLs")
+
+                        if not is_valid_m3u:
+                            # Log what we actually received for debugging
+                            logger.error(f"Invalid M3U content received. First 200 characters: {head_str[:200]!r}")
+
+                            # Try to provide more specific error messages based on content
+                            if '<html' in head_lower or '<!doctype html' in head_lower:
+                                error_msg = f"Server returned HTML page instead of M3U file from URL: {account.server_url}. This usually indicates an error or authentication issue."
+                            elif 'error' in head_lower or 'not found' in head_lower:
+                                error_msg = f"Server returned an error message instead of M3U file from URL: {account.server_url}. Content: {head_str[:100]}"
+                            elif len(head_str.strip()) == 0:
+                                error_msg = f"Server returned completely empty response from URL: {account.server_url}"
+                            else:
+                                error_msg = f"Server provided invalid M3U content from URL: {account.server_url}. Content does not appear to be a valid M3U file."
+                            logger.error(error_msg)
+                            account.status = M3UAccount.Status.ERROR
+                            account.last_message = error_msg
+                            account.save(update_fields=["status", "last_message"])
+                            send_m3u_update(
+                                account.id,
+                                "downloading",
+                                100,
+                                status="error",
+                                error=error_msg,
+                            )
+                            return [], False
+
+                    except UnicodeDecodeError:
+                        with open(temp_path, "rb") as vf:
+                            first_bytes = vf.read(200)
+                        logger.error(f"Non-text content received. First 200 bytes: {first_bytes!r}")
+                        error_msg = f"Server provided non-text content from URL: {account.server_url}. Unable to process as M3U file."
+                        logger.error(error_msg)
+                        account.status = M3UAccount.Status.ERROR
+                        account.last_message = error_msg
+                        account.save(update_fields=["status", "last_message"])
+                        send_m3u_update(
+                            account.id,
+                            "downloading",
+                            100,
+                            status="error",
+                            error=error_msg,
+                        )
+                        return [], False
+
+                    # Validation passed — promote temp file to final path
+                    os.replace(temp_path, file_path)
+
+                    # Final update with 100% progress
+                    dl_size = downloaded / 1024 / 1024
+                    final_msg = f"Download complete. Size: {dl_size:.2f} MB, Time: {time.time() - start_time:.1f}s"
+                    account.last_message = final_msg
+                    account.save(update_fields=["last_message"])
+                    send_m3u_update(account.id, "downloading", 100, message=final_msg)
+
+                finally:
+                    # Clean up temp file on any failure path
+                    if os.path.exists(temp_path):
+                        try:
+                            os.remove(temp_path)
+                        except OSError:
+                            pass
             except requests.exceptions.HTTPError as e:
                 # Handle HTTP errors specifically with more context
                 status_code = e.response.status_code if e.response else "unknown"
@@ -1210,9 +1235,13 @@ def refresh_m3u_groups(account_id, use_cache=False, full_refresh=False, scan_sta
     if not acquire_task_lock("refresh_m3u_account_groups", account_id):
         return f"Task already running for account_id={account_id}.", None
 
+    lock_renewer = TaskLockRenewer("refresh_m3u_account_groups", account_id)
+    lock_renewer.start()
+
     try:
         account = M3UAccount.objects.get(id=account_id, is_active=True)
     except M3UAccount.DoesNotExist:
+        lock_renewer.stop()
         release_task_lock("refresh_m3u_account_groups", account_id)
         return f"M3UAccount with ID={account_id} not found or inactive.", None
 
@@ -1238,6 +1267,7 @@ def refresh_m3u_groups(account_id, use_cache=False, full_refresh=False, scan_sta
             send_m3u_update(
                 account_id, "processing_groups", 100, status="error", error=error_msg
             )
+            lock_renewer.stop()
             release_task_lock("refresh_m3u_account_groups", account_id)
             return error_msg, None
 
@@ -1250,6 +1280,7 @@ def refresh_m3u_groups(account_id, use_cache=False, full_refresh=False, scan_sta
             send_m3u_update(
                 account_id, "processing_groups", 100, status="error", error=error_msg
             )
+            lock_renewer.stop()
             release_task_lock("refresh_m3u_account_groups", account_id)
             return error_msg, None
 
@@ -1359,6 +1390,7 @@ def refresh_m3u_groups(account_id, use_cache=False, full_refresh=False, scan_sta
                                 status="error",
                                 error=error_msg,
                             )
+                            lock_renewer.stop()
                             release_task_lock("refresh_m3u_account_groups", account_id)
                             return error_msg, None
 
@@ -1397,6 +1429,7 @@ def refresh_m3u_groups(account_id, use_cache=False, full_refresh=False, scan_sta
                             status="error",
                             error=error_msg,
                         )
+                        lock_renewer.stop()
                         release_task_lock("refresh_m3u_account_groups", account_id)
                         return error_msg, None
 
@@ -1413,6 +1446,7 @@ def refresh_m3u_groups(account_id, use_cache=False, full_refresh=False, scan_sta
                     status="error",
                     error=error_msg,
                 )
+                lock_renewer.stop()
                 release_task_lock("refresh_m3u_account_groups", account_id)
                 return error_msg, None
         except Exception as e:
@@ -1424,6 +1458,7 @@ def refresh_m3u_groups(account_id, use_cache=False, full_refresh=False, scan_sta
             send_m3u_update(
                 account_id, "processing_groups", 100, status="error", error=error_msg
             )
+            lock_renewer.stop()
             release_task_lock("refresh_m3u_account_groups", account_id)
             return error_msg, None
     else:
@@ -1431,6 +1466,7 @@ def refresh_m3u_groups(account_id, use_cache=False, full_refresh=False, scan_sta
         lines, success = fetch_m3u_lines(account, use_cache)
         if not success:
             # If fetch failed, don't continue processing
+            lock_renewer.stop()
             release_task_lock("refresh_m3u_account_groups", account_id)
             return f"Failed to fetch M3U data for account_id={account_id}.", None
 
@@ -1525,6 +1561,7 @@ def refresh_m3u_groups(account_id, use_cache=False, full_refresh=False, scan_sta
 
     process_groups(account, groups, scan_start_time)
 
+    lock_renewer.stop()
     release_task_lock("refresh_m3u_account_groups", account_id)
 
     if not full_refresh:
@@ -2593,12 +2630,18 @@ def refresh_account_info(profile_id):
 
         release_task_lock("refresh_account_info", profile_id)
         return error_msg
-@shared_task
+@shared_task(time_limit=3600, soft_time_limit=3500)
 def refresh_single_m3u_account(account_id):
     """Splits M3U processing into chunks and dispatches them as parallel tasks."""
     if not acquire_task_lock("refresh_single_m3u_account", account_id):
         return f"Task already running for account_id={account_id}."
 
+    # Keep the lock alive while this long-running task is working.
+    # Without renewal, the 300s lock TTL can expire during large
+    # downloads/parses, allowing duplicate tasks to start.
+    lock_renewer = TaskLockRenewer("refresh_single_m3u_account", account_id)
+    lock_renewer.start()
+
     # Record start time
     refresh_start_timestamp = timezone.now()  # For the cleanup function
     start_time = time.time()  # For tracking elapsed time as float
@@ -3055,6 +3098,8 @@ def refresh_single_m3u_account(account_id):
         account.last_message = f"Error processing M3U: {str(e)}"
         account.save(update_fields=["status", "last_message"])
         raise  # Re-raise the exception for Celery to handle
+    finally:
+        lock_renewer.stop()
 
     release_task_lock("refresh_single_m3u_account", account_id)
 
diff --git a/core/utils.py b/core/utils.py
index bd80782e..46355323 100644
--- a/core/utils.py
+++ b/core/utils.py
@@ -222,6 +222,75 @@ def release_task_lock(task_name, id):
     # Remove the lock
     redis_client.delete(lock_id)
 
+
+class TaskLockRenewer:
+    """Periodically renews a Redis task lock to prevent expiry during long-running tasks.
+
+    Use as a context manager after acquiring a lock:
+
+        if acquire_task_lock("my_task", task_id):
+            with TaskLockRenewer("my_task", task_id):
+                # ... long-running work ...
+            release_task_lock("my_task", task_id)
+
+    A daemon thread extends the lock TTL at regular intervals so that
+    slow downloads or large parsing jobs don't lose their lock mid-operation.
+    """
+
+    def __init__(self, task_name, id, ttl=300, renewal_interval=120):
+        self.task_name = task_name
+        self.id = id
+        self.ttl = ttl
+        self.renewal_interval = renewal_interval
+        self.lock_id = f"task_lock_{task_name}_{id}"
+        self._stop_event = threading.Event()
+        self._thread = None
+
+    def _renew_loop(self):
+        """Background loop that extends the lock TTL until stopped."""
+        while not self._stop_event.wait(self.renewal_interval):
+            try:
+                redis_client = RedisClient.get_client()
+                if redis_client.exists(self.lock_id):
+                    redis_client.expire(self.lock_id, self.ttl)
+                    logger.debug(
+                        f"Renewed lock {self.lock_id} TTL to {self.ttl}s"
+                    )
+                else:
+                    # Lock was deleted externally (e.g. manual release) — stop renewing
+                    logger.warning(
+                        f"Lock {self.lock_id} no longer exists, stopping renewal"
+                    )
+                    break
+            except Exception as e:
+                logger.error(f"Error renewing lock {self.lock_id}: {e}")
+
+    def start(self):
+        """Start the background renewal thread."""
+        self._stop_event.clear()
+        self._thread = threading.Thread(
+            target=self._renew_loop, daemon=True,
+            name=f"lock-renew-{self.task_name}-{self.id}"
+        )
+        self._thread.start()
+        return self
+
+    def stop(self):
+        """Stop the renewal thread."""
+        self._stop_event.set()
+        if self._thread and self._thread.is_alive():
+            self._thread.join(timeout=5)
+        self._thread = None
+
+    def __enter__(self):
+        self.start()
+        return self
+
+    def __exit__(self, exc_type, exc_val, exc_tb):
+        self.stop()
+        return False
+
+
 def send_websocket_update(group_name, event_type, data, collect_garbage=False):
     """
     Standardized function to send WebSocket updates with proper memory management.