Improve error handling and buffering in m3u8 processing

mediaflow_proxy/handlers.py

@@ -3,6 +3,7 @@ import logging
 from urllib.parse import urlparse, parse_qs
 
 import httpx
+import tenacity
 from fastapi import Request, Response, HTTPException
 from starlette.background import BackgroundTask
 
@@ -52,6 +53,8 @@ def handle_exceptions(exception: Exception) -> Response:
     elif isinstance(exception, DownloadError):
         logger.error(f"Error downloading content: {exception}")
         return Response(status_code=exception.status_code, content=str(exception))
+    elif isinstance(exception, tenacity.RetryError):
+        return Response(status_code=502, content="Max retries exceeded while downloading content")
     else:
         logger.exception(f"Internal server error while handling request: {exception}")
         return Response(status_code=502, content=f"Internal server error: {exception}")

mediaflow_proxy/utils/http_utils.py

@@ -33,9 +33,8 @@ class DownloadError(Exception):
 def create_httpx_client(follow_redirects: bool = True, **kwargs) -> httpx.AsyncClient:
     """Creates an HTTPX client with configured proxy routing"""
     mounts = settings.transport_config.get_mounts()
-    client = httpx.AsyncClient(
+    kwargs.setdefault("timeout", settings.transport_config.timeout)
-        mounts=mounts, follow_redirects=follow_redirects, timeout=settings.transport_config.timeout, **kwargs
+    client = httpx.AsyncClient(mounts=mounts, follow_redirects=follow_redirects, **kwargs)
-    )
     return client
 
 
@@ -125,9 +124,12 @@ class Streamer:
             raise DownloadError(
                 e.response.status_code, f"HTTP error {e.response.status_code} while creating streaming response"
             )
+        except httpx.RequestError as e:
+            logger.error(f"Error creating streaming response: {e}")
+            raise DownloadError(502, f"Error creating streaming response: {e}")
         except Exception as e:
             logger.error(f"Error creating streaming response: {e}")
-            raise
+            raise RuntimeError(f"Error creating streaming response: {e}")
 
     async def stream_content(self) -> typing.AsyncGenerator[bytes, None]:
         """

mediaflow_proxy/utils/m3u8_processor.py

@@ -1,4 +1,4 @@
-import logging
+import codecs
 import re
 from typing import AsyncGenerator
 from urllib import parse
@@ -57,54 +57,38 @@ class M3U8Processor:
         Yields:
             str: Processed lines of the m3u8 content.
         """
-        buffer = b""  # Use bytes buffer instead of str
+        buffer = ""  # String buffer for decoded content
+        decoder = codecs.getincrementaldecoder("utf-8")(errors="replace")
 
         # Process the content chunk by chunk
         async for chunk in content_iterator:
             if isinstance(chunk, str):
                 chunk = chunk.encode("utf-8")
 
-            buffer += chunk
+            # Incrementally decode the chunk
+            decoded_chunk = decoder.decode(chunk)
+            buffer += decoded_chunk
 
-            try:
+            # Process complete lines
-                # Try to decode the current buffer with error handling
+            lines = buffer.split("\n")
-                decoded_buffer = buffer.decode("utf-8", errors="replace")
+            if len(lines) > 1:
-                lines = decoded_buffer.split("\n")
+                # Process all complete lines except the last one
+                for line in lines[:-1]:
+                    if line:  # Skip empty lines
+                        processed_line = await self.process_line(line, base_url)
+                        yield processed_line + "\n"
 
-                # If we have at least one line and the buffer ends with newline,
+                # Keep the last line in the buffer (it might be incomplete)
-                # we process all lines including the last one
+                buffer = lines[-1]
-                if len(lines) > 1:
-                    # Process all complete lines except the last one
-                    for line in lines[:-1]:
-                        if line:  # Skip empty lines
-                            processed_line = await self.process_line(line, base_url)
-                            yield processed_line + "\n"
 
-                    # If the buffer ended with a newline, the last line is also complete
+        # Process any remaining data in the buffer plus final bytes
-                    if decoded_buffer.endswith("\n"):
+        final_chunk = decoder.decode(b"", final=True)
-                        if lines[-1]:  # Process the last line if it's not empty
+        if final_chunk:
-                            processed_line = await self.process_line(lines[-1], base_url)
+            buffer += final_chunk
-                            yield processed_line + "\n"
-                        buffer = b""  # Clear the buffer
-                    else:
-                        # Keep the last incomplete line in the buffer
-                        buffer = lines[-1].encode("utf-8")
-            except UnicodeDecodeError:
-                # If we can't decode the buffer, keep accumulating data
-                continue
 
-        # Process any remaining data in the buffer
+        if buffer:  # Process the last line if it's not empty
-        if buffer:
+            processed_line = await self.process_line(buffer, base_url)
-            try:
+            yield processed_line
-                # Final attempt to decode any remaining data
-                remaining = buffer.decode("utf-8", errors="replace")
-                if remaining:
-                    processed_line = await self.process_line(remaining, base_url)
-                    yield processed_line
-            except Exception as e:
-                logging.error(f"Error processing final buffer: {e}")
-                # Yield any remaining content with replacement characters
-                yield await self.process_line(buffer.decode("utf-8", errors="replace"), base_url)
 
     async def process_line(self, line: str, base_url: str) -> str:
         """