diff --git a/CHANGELOG.md b/CHANGELOG.md
index fadc44e8..88f6797a 100644
--- a/CHANGELOG.md
+++ b/CHANGELOG.md
@@ -9,10 +9,22 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0
 
 ### Added
 
+- New Client Buffer proxy setting: new clients joining an active channel are now positioned a configurable number of seconds behind live rather than a fixed chunk count. The start position is determined by wall-clock chunk receive time (stored as a Redis sorted set alongside the buffer), so the buffer depth is consistent in seconds regardless of stream bitrate. Setting the value to `0` starts clients at live with no buffer. Defaults to 2 seconds. Existing chunk-count gating for the first client connecting to a channel is unchanged. The setting is exposed in Settings → Proxy as "New Client Buffer (seconds)".
 - Channel table filter for channels that have stale streams: A new "Has Stale Streams" filter option in the channel table header menu highlights and filters channels containing at least one stale stream. Channels with stale streams are visually distinguished with an orange tint. The filter is mutually exclusive with "Only Empty Channels". - Thanks [@JCBird1012](https://github.com/JCBird1012)
 
 ### Fixed
 
+- TS proxy connection slot leaks and TOCTOU races in stream initialization (Fixes #947) - Thanks [@CodeBormen](https://github.com/CodeBormen)
+  - **TOCTOU race in slot reservation**: `get_stream()` previously used a `GET`→check→`INCR` sequence, allowing concurrent requests to both read the same count below the limit and both reserve a slot, silently exceeding `max_streams`. Replaced with an atomic `INCR`-first pattern: increment unconditionally, check the result, roll back with `DECR` if over capacity.
+  - **Leak on URL generation failure**: `generate_stream_url()` called `get_stream()` (which `INCR`s the counter) but had no cleanup path if subsequent DB lookups or URL construction failed. The post-`get_stream()` block is now wrapped in a `try/except` that calls `release_stream()` on any error.
+  - **Leak on retry-loop timeout**: the retry loop in `stream_ts()` called a bare `get_stream()` on the first failure to classify the error reason. If a slot was available, this `INCR`'d the counter and set Redis keys that were never released when the loop timed out. A `release_stream()` call is now issued before returning 503.
+  - **Leak on `initialize_channel()` failure**: when `initialize_channel()` returned `False`, the connection slot allocated by the preceding `get_stream()` was never released. A `connection_allocated` flag now tracks whether this request performed the `INCR` (fresh initialization vs. joining an existing channel), and `release_stream()` is called guarded by that flag to prevent incorrect decrements when attaching to an already-running channel.
+  - **Safety net for unexpected exceptions**: the outer `except` in `stream_ts()` now checks `connection_allocated` and calls `release_stream()` as a last-resort cleanup for any unhandled exception that escapes before the channel is handed off to the stream lifecycle.
+  - **`release_stream()` now returns `bool`** and adds a metadata-hash fallback: if the primary `channel_stream` / `stream_profile` Redis keys have already been cleaned up by the proxy, it recovers `stream_id` and `profile_id` from the channel's metadata hash and clears those fields atomically to prevent duplicate `DECR`s on repeated calls.
+  - **`update_stream_profile()` uses a Redis pipeline** for the old-profile DECR + key update + new-profile INCR sequence, preventing counter drift if the process crashes between operations.
+  - **`stream_generator._cleanup()`** now falls back to `Stream.objects.get()` when the channel UUID resolves to a preview flow rather than a normal channel, rather than silently skipping the slot release.
+  - **VOD `cleanup_persistent_connection()`** fallback DECR is now conditional: it only decrements the profile counter when the connection tracking key had already expired by TTL (i.e., `remove_connection()` would have skipped the DECR), preventing double-decrements when the key is still present.
+- TS proxy client stream lag recovery now only bumps clients forward when their next required chunk has genuinely expired from Redis (TTL), rather than unconditionally jumping if they fell more than 50 chunks behind. Clients are repositioned to the oldest available chunk (minimum data loss) using an atomic server-side Lua binary search, falling back to near the buffer head if nothing is available.
 - TS proxy streams dying after 30–200 seconds in multi-worker uWSGI/Celery deployments, caused by three interrelated bugs. (Fixes #992, #980) - Thanks [@PFalko](https://github.com/PFalko)
   - **Double ProxyServer instantiation**: `ProxyConfig.ready()` called `TSProxyServer()` directly while `TSProxyConfig.ready()` also called `TSProxyServer.get_instance()`, creating two instances per worker — each with its own cleanup thread. The orphaned thread could not extend ownership because it had no entries in `stream_managers`. Fixed by using `TSProxyServer.get_instance()` in `ProxyConfig.ready()`.
   - **`flushdb()` on every Redis client init**: `RedisClient.get_client()` called `client.flushdb()` whenever `_client` was `None`. Celery autoscale (`--autoscale=6,1`) spawning new workers mid-stream triggered this path, nuking all Redis keys including active ownership keys, client records, and channel metadata. Removed the `flushdb()` call entirely.
diff --git a/apps/channels/models.py b/apps/channels/models.py
index ec5e135b..217a92c0 100644
--- a/apps/channels/models.py
+++ b/apps/channels/models.py
@@ -3,6 +3,8 @@ from django.core.exceptions import ValidationError
 from django.conf import settings
 from core.models import StreamProfile, CoreSettings
 from core.utils import RedisClient
+from apps.proxy.ts_proxy.redis_keys import RedisKeys
+from apps.proxy.ts_proxy.constants import ChannelMetadataField
 import logging
 import uuid
 from datetime import datetime
@@ -204,7 +206,10 @@ class Stream(models.Model):
 
     def get_stream(self):
         """
-        Finds an available stream for the requested channel and returns the selected stream and profile.
+        Finds an available profile for this stream and reserves a connection slot.
+
+        Returns:
+            Tuple[Optional[int], Optional[int], Optional[str]]: (stream_id, profile_id, error_reason)
         """
         redis_client = RedisClient.get_client()
         profile_id = redis_client.get(f"stream_profile:{self.id}")
@@ -226,33 +231,32 @@ class Stream(models.Model):
             if profile.is_active == False:
                 continue
 
-            profile_connections_key = f"profile_connections:{profile.id}"
-            current_connections = int(redis_client.get(profile_connections_key) or 0)
+            # Atomic slot reservation: INCR first, check, rollback if over capacity
+            if profile.max_streams == 0:
+                reserved = True
+            else:
+                profile_connections_key = f"profile_connections:{profile.id}"
+                new_count = redis_client.incr(profile_connections_key)
+                if new_count <= profile.max_streams:
+                    reserved = True
+                else:
+                    redis_client.decr(profile_connections_key)
+                    reserved = False
 
-            # Check if profile has available slots (or unlimited connections)
-            if profile.max_streams == 0 or current_connections < profile.max_streams:
-                # Start a new stream
+            if reserved:
                 redis_client.set(f"channel_stream:{self.id}", self.id)
-                redis_client.set(
-                    f"stream_profile:{self.id}", profile.id
-                )  # Store only the matched profile
+                redis_client.set(f"stream_profile:{self.id}", profile.id)
+                return self.id, profile.id, None
 
-                # Increment connection count for profiles with limits
-                if profile.max_streams > 0:
-                    redis_client.incr(profile_connections_key)
-
-                return (
-                    self.id,
-                    profile.id,
-                    None,
-                )  # Return newly assigned stream and matched profile
-
-        # 4. No available streams
-        return None, None, None
+        return None, None, "All active M3U profiles have reached maximum connection limits"
 
     def release_stream(self):
         """
         Called when a stream is finished to release the lock.
+
+        Returns:
+            bool: True if stream was successfully released, False if
+                  no profile info could be found for cleanup.
         """
         redis_client = RedisClient.get_client()
 
@@ -260,14 +264,17 @@ class Stream(models.Model):
         # Get the matched profile for cleanup
         profile_id = redis_client.get(f"stream_profile:{stream_id}")
         if not profile_id:
-            logger.debug("Invalid profile ID pulled from stream index")
-            return
+            logger.debug(
+                f"Stream {stream_id}: no profile found in "
+                f"stream_profile:{stream_id}"
+            )
+            return False
 
         redis_client.delete(f"stream_profile:{stream_id}")  # Remove profile association
 
         profile_id = int(profile_id)
         logger.debug(
-            f"Found profile ID {profile_id} associated with stream {stream_id}"
+            f"Stream {stream_id}: found profile_id={profile_id}"
         )
 
         profile_connections_key = f"profile_connections:{profile_id}"
@@ -277,6 +284,8 @@ class Stream(models.Model):
         if current_count > 0:
             redis_client.decr(profile_connections_key)
 
+        return True
+
 
 class ChannelManager(models.Manager):
     def active(self):
@@ -391,6 +400,38 @@ class Channel(models.Model):
 
         return stream_profile
 
+    def _check_and_reserve_profile_slot(self, profile, redis_client):
+        """
+        Atomically check and reserve a connection slot for the given profile.
+
+        Uses an INCR-first-then-check pattern to eliminate the TOCTOU race
+        condition where separate GET + check + INCR operations could allow
+        concurrent requests to both pass the capacity check.
+
+        For profiles with max_streams=0 (unlimited), no reservation is needed.
+
+        Args:
+            profile: M3UAccountProfile instance
+            redis_client: Redis client instance
+
+        Returns:
+            tuple: (reserved: bool, current_count: int)
+        """
+        if profile.max_streams == 0:
+            return (True, 0)
+
+        profile_connections_key = f"profile_connections:{profile.id}"
+
+        # Atomically increment first — this is a single Redis command
+        new_count = redis_client.incr(profile_connections_key)
+
+        if new_count <= profile.max_streams:
+            return (True, new_count)
+
+        # Over capacity — roll back the increment
+        redis_client.decr(profile_connections_key)
+        return (False, new_count - 1)
+
     def get_stream(self):
         """
         Finds an available stream for the requested channel and returns the selected stream and profile.
@@ -456,24 +497,16 @@ class Channel(models.Model):
             for profile in profiles:
                 has_active_profiles = True
 
-                profile_connections_key = f"profile_connections:{profile.id}"
-                current_connections = int(
-                    redis_client.get(profile_connections_key) or 0
+                # Atomically check and reserve a slot (INCR-first pattern)
+                reserved, current_count = self._check_and_reserve_profile_slot(
+                    profile, redis_client
                 )
 
-                # Check if profile has available slots (or unlimited connections)
-                if (
-                    profile.max_streams == 0
-                    or current_connections < profile.max_streams
-                ):
-                    # Start a new stream
+                if reserved:
+                    # Slot reserved — assign stream to this channel
                     redis_client.set(f"channel_stream:{self.id}", stream.id)
                     redis_client.set(f"stream_profile:{stream.id}", profile.id)
 
-                    # Increment connection count for profiles with limits
-                    if profile.max_streams > 0:
-                        redis_client.incr(profile_connections_key)
-
                     return (
                         stream.id,
                         profile.id,
@@ -483,7 +516,8 @@ class Channel(models.Model):
                     # This profile is at max connections
                     has_streams_but_maxed_out = True
                     logger.debug(
-                        f"Profile {profile.id} at max connections: {current_connections}/{profile.max_streams}"
+                        f"Profile {profile.id} at max connections: "
+                        f"{current_count}/{profile.max_streams}"
                     )
 
         # No available streams - determine specific reason
@@ -499,32 +533,93 @@ class Channel(models.Model):
     def release_stream(self):
         """
         Called when a stream is finished to release the lock.
+
+        Returns:
+            bool: True if stream was successfully released, False if
+                  no stream/profile info could be found for cleanup.
         """
         redis_client = RedisClient.get_client()
 
         stream_id = redis_client.get(f"channel_stream:{self.id}")
         if not stream_id:
-            logger.debug("Invalid stream ID pulled from channel index")
-            return
+            # Primary key missing — try metadata hash fallback.
+            # The proxy may have already cleaned up channel_stream/stream_profile
+            # keys, but the metadata hash can still have the stream_id and profile.
+            metadata_key = RedisKeys.channel_metadata(str(self.uuid))
+            meta_stream_id = redis_client.hget(
+                metadata_key, ChannelMetadataField.STREAM_ID
+            )
+            meta_profile_id = redis_client.hget(
+                metadata_key, ChannelMetadataField.M3U_PROFILE
+            )
+
+            if meta_stream_id and meta_profile_id:
+                stream_id = int(meta_stream_id)
+                profile_id = int(meta_profile_id)
+                logger.debug(
+                    f"Channel {self.uuid}: recovered stream_id={stream_id}, "
+                    f"profile_id={profile_id} from metadata fallback"
+                )
+                # Clean up any remaining keys
+                redis_client.delete(f"channel_stream:{self.id}")
+                redis_client.delete(f"stream_profile:{stream_id}")
+
+                # Clear metadata fields so duplicate release_stream() calls
+                # won't find them and DECR again
+                redis_client.hdel(
+                    metadata_key,
+                    ChannelMetadataField.STREAM_ID,
+                    ChannelMetadataField.M3U_PROFILE,
+                )
+
+                profile_connections_key = f"profile_connections:{profile_id}"
+                current_count = int(
+                    redis_client.get(profile_connections_key) or 0
+                )
+                if current_count > 0:
+                    redis_client.decr(profile_connections_key)
+                return True
+
+            logger.debug(
+                f"Channel {self.uuid}: no stream info found in primary keys "
+                f"or metadata fallback"
+            )
+            return False
 
         redis_client.delete(f"channel_stream:{self.id}")  # Remove active stream
 
         stream_id = int(stream_id)
         logger.debug(
-            f"Found stream ID {stream_id} associated with channel stream {self.id}"
+            f"Channel {self.uuid}: found stream_id={stream_id} for "
+            f"channel_stream:{self.id}"
         )
 
         # Get the matched profile for cleanup
         profile_id = redis_client.get(f"stream_profile:{stream_id}")
-        if not profile_id:
-            logger.debug("Invalid profile ID pulled from stream index")
-            return
-
-        redis_client.delete(f"stream_profile:{stream_id}")  # Remove profile association
-
-        profile_id = int(profile_id)
+        if profile_id:
+            redis_client.delete(f"stream_profile:{stream_id}")  # Remove profile association
+            profile_id = int(profile_id)
+        else:
+            # stream_profile key missing — try metadata hash fallback
+            metadata_key = RedisKeys.channel_metadata(str(self.uuid))
+            meta_profile_id = redis_client.hget(
+                metadata_key, ChannelMetadataField.M3U_PROFILE
+            )
+            if meta_profile_id:
+                profile_id = int(meta_profile_id)
+                logger.debug(
+                    f"Channel {self.uuid}: recovered profile_id={profile_id} "
+                    f"from metadata fallback (stream_profile:{stream_id} was missing)"
+                )
+            else:
+                logger.warning(
+                    f"Channel {self.uuid}: no profile found for "
+                    f"stream_profile:{stream_id} or in metadata fallback"
+                )
+                return False
         logger.debug(
-            f"Found profile ID {profile_id} associated with stream {stream_id}"
+            f"Channel {self.uuid}: found profile_id={profile_id} for "
+            f"stream {stream_id}"
         )
 
         profile_connections_key = f"profile_connections:{profile_id}"
@@ -534,6 +629,18 @@ class Channel(models.Model):
         if current_count > 0:
             redis_client.decr(profile_connections_key)
 
+        # Clear metadata fields so duplicate release_stream() calls
+        # (e.g. from _clean_redis_keys or ChannelService.stop_channel)
+        # won't find them via fallback and DECR again
+        metadata_key = RedisKeys.channel_metadata(str(self.uuid))
+        redis_client.hdel(
+            metadata_key,
+            ChannelMetadataField.STREAM_ID,
+            ChannelMetadataField.M3U_PROFILE,
+        )
+
+        return True
+
     def update_stream_profile(self, new_profile_id):
         """
         Updates the profile for the current stream and adjusts connection counts.
@@ -566,18 +673,18 @@ class Channel(models.Model):
         if current_profile_id == new_profile_id:
             return True
 
-        # Decrement connection count for old profile
+        # Use pipeline for atomic profile switch to prevent counter drift
+        # if an exception occurs between DECR and INCR
         old_profile_connections_key = f"profile_connections:{current_profile_id}"
-        old_count = int(redis_client.get(old_profile_connections_key) or 0)
-        if old_count > 0:
-            redis_client.decr(old_profile_connections_key)
-
-        # Update the profile mapping
-        redis_client.set(f"stream_profile:{stream_id}", new_profile_id)
-
-        # Increment connection count for new profile
         new_profile_connections_key = f"profile_connections:{new_profile_id}"
-        redis_client.incr(new_profile_connections_key)
+        old_count = int(redis_client.get(old_profile_connections_key) or 0)
+
+        pipe = redis_client.pipeline()
+        if old_count > 0:
+            pipe.decr(old_profile_connections_key)
+        pipe.set(f"stream_profile:{stream_id}", new_profile_id)
+        pipe.incr(new_profile_connections_key)
+        pipe.execute()
         logger.info(
             f"Updated stream {stream_id} profile from {current_profile_id} to {new_profile_id}"
         )
diff --git a/apps/proxy/config.py b/apps/proxy/config.py
index 3b1ce967..dbdd8278 100644
--- a/apps/proxy/config.py
+++ b/apps/proxy/config.py
@@ -43,6 +43,7 @@ class BaseConfig:
                 "redis_chunk_ttl": 60,
                 "channel_shutdown_delay": 0,
                 "channel_init_grace_period": 5,
+                "new_client_behind_seconds": 2,
             }
 
         finally:
@@ -81,6 +82,7 @@ class TSConfig(BaseConfig):
     # Buffer settings
     INITIAL_BEHIND_CHUNKS = 4  # How many chunks behind to start a client (4 chunks = ~1MB)
     CHUNK_BATCH_SIZE = 5       # How many chunks to fetch in one batch
+    NEW_CLIENT_BEHIND_SECONDS = 2  # Start new clients this many seconds behind live (0 = start at live)
     KEEPALIVE_INTERVAL = 0.5   # Seconds between keepalive packets when at buffer head
     # Chunk read timeout
     CHUNK_TIMEOUT = 5        # Seconds to wait for each chunk read
diff --git a/apps/proxy/ts_proxy/config_helper.py b/apps/proxy/ts_proxy/config_helper.py
index d7d33558..f0f7a5b7 100644
--- a/apps/proxy/ts_proxy/config_helper.py
+++ b/apps/proxy/ts_proxy/config_helper.py
@@ -41,6 +41,15 @@ class ConfigHelper:
         """Get number of chunks to start behind"""
         return ConfigHelper.get('INITIAL_BEHIND_CHUNKS', 4)
 
+    @staticmethod
+    def new_client_behind_seconds():
+        """Get number of seconds behind live to start new clients.
+        0 means start at live (buffer head).
+        Loaded from DB proxy_settings so users can change it at runtime."""
+        from apps.proxy.config import TSConfig
+        settings = TSConfig.get_proxy_settings()
+        return settings.get('new_client_behind_seconds', 2)
+
     @staticmethod
     def keepalive_interval():
         """Get keepalive interval in seconds"""
diff --git a/apps/proxy/ts_proxy/redis_keys.py b/apps/proxy/ts_proxy/redis_keys.py
index 22b02648..29846f98 100644
--- a/apps/proxy/ts_proxy/redis_keys.py
+++ b/apps/proxy/ts_proxy/redis_keys.py
@@ -79,6 +79,12 @@ class RedisKeys:
         """Key for worker heartbeat"""
         return f"ts_proxy:worker:{worker_id}:heartbeat"
 
+    @staticmethod
+    def chunk_timestamps(channel_id):
+        """Sorted set mapping chunk receive-timestamps (score) to chunk indices (member).
+        Used for time-based client positioning."""
+        return f"ts_proxy:channel:{channel_id}:buffer:chunk_timestamps"
+
     @staticmethod
     def transcode_active(channel_id):
         """Key indicating active transcode process"""
diff --git a/apps/proxy/ts_proxy/server.py b/apps/proxy/ts_proxy/server.py
index 609a3000..e1e2c545 100644
--- a/apps/proxy/ts_proxy/server.py
+++ b/apps/proxy/ts_proxy/server.py
@@ -796,7 +796,10 @@ class ProxyServer:
                 if self.redis_client.exists(key):
                     # Found orphaned keys without metadata - clean them up
                     logger.warning(f"Found orphaned keys for channel {channel_id} without metadata - cleaning up")
-                    self._clean_redis_keys(channel_id)
+                    try:
+                        self._clean_redis_keys(channel_id)
+                    except Exception as e:
+                        logger.error(f"Error cleaning redis keys for channel {channel_id}: {e}")
                     return False
 
         return False
@@ -818,13 +821,17 @@ class ProxyServer:
             # Force release resources in the Channel model
             try:
                 channel = Channel.objects.get(uuid=channel_id)
-                channel.release_stream()
-                logger.info(f"Released stream allocation for zombie channel {channel_id}")
+                if not channel.release_stream():
+                    logger.warning(f"Failed to release stream for zombie channel {channel_id}")
+                else:
+                    logger.info(f"Released stream allocation for zombie channel {channel_id}")
             except Exception as e:
                 try:
                     stream = Stream.objects.get(stream_hash=channel_id)
-                    stream.release_stream()
-                    logger.info(f"Released stream allocation for zombie channel {channel_id}")
+                    if not stream.release_stream():
+                        logger.warning(f"Failed to release stream for zombie channel {channel_id}")
+                    else:
+                        logger.info(f"Released stream allocation for zombie channel {channel_id}")
                 except Exception as e:
                     logger.error(f"Error releasing stream for zombie channel {channel_id}: {e}")
 
@@ -1396,10 +1403,15 @@ class ProxyServer:
         # Release the channel, stream, and profile keys from the channel
         try:
             channel = Channel.objects.get(uuid=channel_id)
-            channel.release_stream()
-        except:
-            stream = Stream.objects.get(stream_hash=channel_id)
-            stream.release_stream()
+            if not channel.release_stream():
+                logger.debug(f"Channel {channel_id}: release_stream found no keys to clean")
+        except (Channel.DoesNotExist, Exception):
+            try:
+                stream = Stream.objects.get(stream_hash=channel_id)
+                if not stream.release_stream():
+                    logger.debug(f"Stream {channel_id}: release_stream found no keys to clean")
+            except (Stream.DoesNotExist, Exception):
+                logger.debug(f"No Channel or Stream found for {channel_id}")
 
         if not self.redis_client:
             return 0
diff --git a/apps/proxy/ts_proxy/services/channel_service.py b/apps/proxy/ts_proxy/services/channel_service.py
index 4c4a73ac..e765ece3 100644
--- a/apps/proxy/ts_proxy/services/channel_service.py
+++ b/apps/proxy/ts_proxy/services/channel_service.py
@@ -265,18 +265,23 @@ class ChannelService:
         # Release the channel in the channel model if applicable
         try:
             channel = Channel.objects.get(uuid=channel_id)
-            channel.release_stream()
-            logger.info(f"Released channel {channel_id} stream allocation")
-            model_released = True
-        except Channel.DoesNotExist:
+            model_released = channel.release_stream()
+            if model_released:
+                logger.info(f"Released channel {channel_id} stream allocation")
+            else:
+                logger.warning(f"Channel {channel_id}: release_stream found no keys to clean")
+        except (Channel.DoesNotExist, Exception):
             logger.warning(f"Could not find Channel model for UUID {channel_id}, attempting stream hash")
-            stream = Stream.objects.get(stream_hash=channel_id)
-            stream.release_stream()
-            logger.info(f"Released stream {channel_id} stream allocation")
-            model_released = True
-        except Exception as e:
-            logger.error(f"Error releasing channel stream: {e}")
-            model_released = False
+            try:
+                stream = Stream.objects.get(stream_hash=channel_id)
+                model_released = stream.release_stream()
+                if model_released:
+                    logger.info(f"Released stream {channel_id} stream allocation")
+                else:
+                    logger.warning(f"Stream {channel_id}: release_stream found no keys to clean")
+            except (Stream.DoesNotExist, Exception) as e:
+                logger.error(f"No Channel or Stream found for {channel_id}: {e}")
+                model_released = False
 
         return {
             'status': 'success',
diff --git a/apps/proxy/ts_proxy/stream_buffer.py b/apps/proxy/ts_proxy/stream_buffer.py
index 85feb5dd..111468bf 100644
--- a/apps/proxy/ts_proxy/stream_buffer.py
+++ b/apps/proxy/ts_proxy/stream_buffer.py
@@ -45,6 +45,22 @@ class StreamBuffer:
         self._write_buffer = bytearray()
         self.target_chunk_size = ConfigHelper.get('BUFFER_CHUNK_SIZE', TS_PACKET_SIZE * 5644)  # ~1MB default
 
+        # Sorted-set key for chunk receive-timestamps (time-based positioning)
+        self.chunk_timestamps_key = RedisKeys.chunk_timestamps(channel_id) if channel_id else ""
+
+        # Register Lua scripts once — subsequent calls use EVALSHA (just the
+        # SHA hash) instead of sending the full script text on every invocation.
+        if self.redis_client:
+            self._find_oldest_chunk_sha = self.redis_client.register_script(
+                self._FIND_OLDEST_CHUNK_LUA
+            )
+            self._find_chunk_by_time_sha = self.redis_client.register_script(
+                self._FIND_CHUNK_BY_TIME_LUA
+            )
+        else:
+            self._find_oldest_chunk_sha = None
+            self._find_chunk_by_time_sha = None
+
         # Track timers for proper cleanup
         self.stopping = False
         self.fill_timers = []
@@ -92,6 +108,14 @@ class StreamBuffer:
                         chunk_key = RedisKeys.buffer_chunk(self.channel_id, chunk_index)
                         self.redis_client.setex(chunk_key, self.chunk_ttl, bytes(chunk_data))
 
+                        # Record receive timestamp for time-based client positioning
+                        if self.chunk_timestamps_key:
+                            now = time.time()
+                            self.redis_client.zadd(self.chunk_timestamps_key, {str(chunk_index): now})
+                            # Prune entries whose chunks have expired from Redis
+                            self.redis_client.zremrangebyscore(self.chunk_timestamps_key, '-inf', now - self.chunk_ttl)
+                            self.redis_client.expire(self.chunk_timestamps_key, self.chunk_ttl)
+
                         # Update local tracking
                         self.index = chunk_index
                         writes_done += 1
@@ -275,6 +299,13 @@ class StreamBuffer:
                 if hasattr(self, '_partial_packet'):
                     self._partial_packet = bytearray()
 
+            # Clean up the chunk timestamps sorted set
+            if self.redis_client and self.chunk_timestamps_key:
+                try:
+                    self.redis_client.delete(self.chunk_timestamps_key)
+                except Exception as e:
+                    logger.error(f"Error deleting chunk timestamps key: {e}")
+
         except Exception as e:
             logger.error(f"Error during buffer stop: {e}")
 
@@ -328,6 +359,146 @@ class StreamBuffer:
 
         return chunks, client_index + chunk_count
 
+    # Lua script that runs an atomic binary search on the Redis server.
+    # Chunks expire in FIFO order (same TTL, sequential writes), so the
+    # alive range is contiguous: [oldest_surviving .. buffer_head].
+    # Binary search finds the boundary in O(log N) EXISTS calls with zero
+    # round-trips between steps and no TOCTOU races (Lua scripts are atomic).
+    #
+    # ARGV[1] = key prefix  (e.g. "ts_proxy:channel:<id>:buffer:chunk:")
+    # ARGV[2] = low index   (client_index + 1, first chunk the client needs)
+    # ARGV[3] = high index  (buffer head, most recent chunk)
+    #
+    # Returns: the index of the oldest existing chunk, or -1 if none exist.
+    _FIND_OLDEST_CHUNK_LUA = """
+    local prefix = ARGV[1]
+    local low    = tonumber(ARGV[2])
+    local high   = tonumber(ARGV[3])
+
+    if redis.call('EXISTS', prefix .. high) == 0 then
+        return -1
+    end
+
+    local result = high
+    while low <= high do
+        local mid = math.floor((low + high) / 2)
+        if redis.call('EXISTS', prefix .. mid) == 1 then
+            result = mid
+            high = mid - 1
+        else
+            low = mid + 1
+        end
+    end
+    return result
+    """
+
+    def find_oldest_available_chunk(self, client_index):
+        """Find the oldest (lowest-index) chunk that still exists in Redis.
+
+        Executes an atomic Lua binary search on the Redis server — one
+        round-trip, ~log2(N) EXISTS calls, no TOCTOU between steps.
+
+        The actual read attempt (get_optimized_client_data) is what
+        authoritatively detects expiration; this method is best-effort
+        positioning that self-corrects on the next iteration if the found
+        chunk also expires before the client can read it.
+
+        Args:
+            client_index: The client's current local_index (last consumed chunk).
+
+        Returns:
+            int or None: The local_index value the client should jump to
+                         (one before the first available chunk), or None if no
+                         chunks are available at all.
+        """
+        if not self.redis_client:
+            return None
+
+        low = client_index + 1   # First chunk the client needs
+        high = self.index        # Latest chunk written
+
+        if low > high:
+            return None
+
+        try:
+            # Uses EVALSHA under the hood — sends only the SHA hash,
+            # not the full script text, on every call after the first.
+            result = self._find_oldest_chunk_sha(
+                args=[
+                    RedisKeys.buffer_chunk_prefix(self.channel_id),
+                    low,
+                    high,
+                ],
+            )
+
+            if result == -1:
+                return None
+
+            # Return result - 1 so local_index points to one before the
+            # first available chunk (matching the "last consumed" convention).
+            return int(result) - 1
+
+        except Exception as e:
+            logger.error(f"Error running find_oldest_chunk Lua script for channel {self.channel_id}: {e}")
+            return None
+
+    # ------------------------------------------------------------------
+    # Lua script: atomic reverse-scan of the chunk_timestamps sorted set.
+    # Finds the chunk whose receive-timestamp is closest to (but <=) a
+    # target wall-clock time.  Returns the chunk index or -1.
+    #
+    # KEYS[1] = chunk_timestamps sorted-set key
+    # ARGV[1] = target timestamp  (time.time() - desired_seconds_behind)
+    # ------------------------------------------------------------------
+    _FIND_CHUNK_BY_TIME_LUA = """
+    local ts_key  = KEYS[1]
+    local target  = tonumber(ARGV[1])
+
+    -- ZREVRANGEBYSCORE returns members with score <= target, highest first.
+    local result = redis.call('ZREVRANGEBYSCORE', ts_key, target, '-inf', 'LIMIT', 0, 1)
+    if #result == 0 then
+        return -1
+    end
+    return tonumber(result[1])
+    """
+
+    def find_chunk_index_by_time(self, seconds_behind):
+        """Find the chunk index that was received approximately *seconds_behind*
+        seconds ago.
+
+        Uses an atomic Lua script against the chunk_timestamps sorted set so
+        no data can expire between the lookup and the read.
+
+        Returns:
+            int or None: The chunk index to position the client at (this is
+                         the *last consumed* convention, so the next read
+                         starts at index+1).  None if no suitable chunk
+                         exists.
+        """
+        if not self.redis_client or not self.chunk_timestamps_key:
+            return None
+
+        target_time = time.time() - seconds_behind
+
+        try:
+            result = self._find_chunk_by_time_sha(
+                keys=[self.chunk_timestamps_key],
+                args=[target_time],
+            )
+            if result is None or int(result) == -1:
+                # No chunk old enough — fall back to the oldest available chunk
+                oldest = self.redis_client.zrange(self.chunk_timestamps_key, 0, 0)
+                if oldest:
+                    return max(0, int(oldest[0]) - 1)  # "last consumed" convention
+                return None
+
+            # Return index - 1 so next read starts at that chunk
+            return max(0, int(result) - 1)
+
+        except Exception as e:
+            logger.error(f"Error in find_chunk_index_by_time for channel {self.channel_id}: {e}")
+            return None
+
     # Add a new method to safely create timers
     def schedule_timer(self, delay, callback, *args, **kwargs):
         """Schedule a timer and track it for proper cleanup"""
diff --git a/apps/proxy/ts_proxy/stream_generator.py b/apps/proxy/ts_proxy/stream_generator.py
index 80329a46..f0174616 100644
--- a/apps/proxy/ts_proxy/stream_generator.py
+++ b/apps/proxy/ts_proxy/stream_generator.py
@@ -8,7 +8,7 @@ import logging
 import threading
 import gevent  # Add this import at the top of your file
 from apps.proxy.config import TSConfig as Config
-from apps.channels.models import Channel
+from apps.channels.models import Channel, Stream
 from core.utils import log_system_event
 from .server import ProxyServer
 from .utils import create_ts_packet, get_logger
@@ -199,10 +199,47 @@ class StreamGenerator:
             logger.error(f"[{self.client_id}] No buffer found for channel {self.channel_id}")
             return False
 
-        # Client state tracking - use config for initial position
-        initial_behind = ConfigHelper.initial_behind_chunks()
+        # Client state tracking — determine start position
+        # When behind_seconds > 0, use time-based positioning to start
+        # the client that many seconds behind live.
+        # When behind_seconds == 0, start at live (buffer head).
+        behind_seconds = ConfigHelper.new_client_behind_seconds()
         current_buffer_index = buffer.index
-        self.local_index = max(0, current_buffer_index - initial_behind)
+
+        if behind_seconds > 0:
+            time_index = buffer.find_chunk_index_by_time(behind_seconds)
+            if time_index is not None:
+                self.local_index = max(0, time_index)
+                logger.info(
+                    f"[{self.client_id}] Time-based positioning: "
+                    f"{behind_seconds}s behind -> index {self.local_index} "
+                    f"(buffer head at {current_buffer_index})"
+                )
+            else:
+                # Not enough buffer for the requested time — start as far
+                # back as possible (oldest available chunk).
+                oldest = buffer.find_oldest_available_chunk(0)
+                if oldest is not None:
+                    self.local_index = max(0, oldest)
+                    logger.info(
+                        f"[{self.client_id}] Buffer shorter than {behind_seconds}s, "
+                        f"starting at oldest available chunk {self.local_index} "
+                        f"(buffer head at {current_buffer_index})"
+                    )
+                else:
+                    # No timestamp data at all — start at live
+                    self.local_index = current_buffer_index
+                    logger.info(
+                        f"[{self.client_id}] No timestamp data, starting at live: "
+                        f"index {self.local_index} (buffer head at {current_buffer_index})"
+                    )
+        else:
+            # 0 = start at live (buffer head)
+            self.local_index = current_buffer_index
+            logger.info(
+                f"[{self.client_id}] Starting at live (behind_seconds=0): "
+                f"index {self.local_index} (buffer head at {current_buffer_index})"
+            )
 
         # Store important objects as instance variables
         self.proxy_server = proxy_server
@@ -238,17 +275,35 @@ class StreamGenerator:
                 self.empty_reads += 1
                 self.consecutive_empty += 1
 
-                # Check if we're too far behind (chunks expired from Redis)
+                # We got no data despite being behind the buffer head.
+                # The read itself is the authoritative signal — no separate
+                # existence check needed, avoiding TOCTOU races with Redis TTL.
                 chunks_behind = self.buffer.index - self.local_index
-                if chunks_behind > 50:  # If more than 50 chunks behind, jump forward
-                    # Calculate new position: stay a few chunks behind current buffer
-                    initial_behind = ConfigHelper.initial_behind_chunks()
-                    new_index = max(self.local_index, self.buffer.index - initial_behind)
+                if chunks_behind > 0:
+                    # Next chunk has expired — find the oldest chunk still in Redis
+                    new_index = self.buffer.find_oldest_available_chunk(self.local_index)
 
-                    logger.warning(f"[{self.client_id}] Client too far behind ({chunks_behind} chunks), jumping from {self.local_index} to {new_index}")
-                    self.local_index = new_index
-                    self.consecutive_empty = 0  # Reset since we're repositioning
-                    continue  # Try again immediately with new position
+                    if new_index is not None:
+                        skipped = new_index - self.local_index
+                        logger.warning(
+                            f"[{self.client_id}] Next chunk expired (index {self.local_index + 1}), "
+                            f"jumping to oldest available: {new_index + 1} "
+                            f"(skipped {skipped} chunks, buffer head at {self.buffer.index})"
+                        )
+                        self.local_index = new_index
+                    else:
+                        # No chunks available at all — jump to near the buffer head
+                        initial_behind = ConfigHelper.initial_behind_chunks()
+                        new_index = max(self.local_index, self.buffer.index - initial_behind)
+                        logger.warning(
+                            f"[{self.client_id}] No chunks available in buffer, "
+                            f"jumping to near buffer head: {new_index} "
+                            f"(buffer head at {self.buffer.index})"
+                        )
+                        self.local_index = new_index
+
+                    self.consecutive_empty = 0
+                    continue  # Retry immediately with the new position
 
                 if self._should_send_keepalive(self.local_index):
                     keepalive_packet = create_ts_packet('keepalive')
@@ -486,11 +541,16 @@ class StreamGenerator:
                             # Only the last client or owner should release the stream
                             if client_count <= 1 and proxy_server.am_i_owner(self.channel_id):
                                 try:
-                                    # Get the channel by UUID
-                                    channel = Channel.objects.get(uuid=self.channel_id)
-                                    channel.release_stream()
-                                    stream_released = True
-                                    logger.debug(f"[{self.client_id}] Released stream for channel {self.channel_id}")
+                                    # Try Channel first (normal flow), fall back to Stream (preview flow)
+                                    try:
+                                        obj = Channel.objects.get(uuid=self.channel_id)
+                                    except (Channel.DoesNotExist, Exception):
+                                        obj = Stream.objects.get(stream_hash=self.channel_id)
+                                    stream_released = obj.release_stream()
+                                    if stream_released:
+                                        logger.debug(f"[{self.client_id}] Released stream for channel {self.channel_id}")
+                                    else:
+                                        logger.warning(f"[{self.client_id}] release_stream found no keys for channel {self.channel_id}")
                                 except Exception as e:
                                     logger.error(f"[{self.client_id}] Error releasing stream for channel {self.channel_id}: {e}")
             except Exception as e:
@@ -519,8 +579,7 @@ class StreamGenerator:
                 logger.error(f"Could not log client disconnect event: {e}")
 
             # Schedule channel shutdown if no clients left
-            if not stream_released:  # Only if we haven't already released the stream
-                self._schedule_channel_shutdown_if_needed(local_clients)
+            self._schedule_channel_shutdown_if_needed(local_clients)
 
     def _schedule_channel_shutdown_if_needed(self, local_clients):
         """
diff --git a/apps/proxy/ts_proxy/url_utils.py b/apps/proxy/ts_proxy/url_utils.py
index 8b467b7f..fe06ad48 100644
--- a/apps/proxy/ts_proxy/url_utils.py
+++ b/apps/proxy/ts_proxy/url_utils.py
@@ -129,39 +129,42 @@ def generate_stream_url(channel_id: str) -> Tuple[str, str, bool, Optional[int]]
             logger.error(f"No stream available for channel {channel_id}: {error_reason}")
             return None, None, False, None
 
-        # Look up the Stream and Profile objects
+        # get_stream() allocated a connection slot - ensure it's released on any error
         try:
+            # Look up the Stream and Profile objects
             stream = Stream.objects.get(id=stream_id)
             profile = M3UAccountProfile.objects.get(id=profile_id)
-        except (Stream.DoesNotExist, M3UAccountProfile.DoesNotExist) as e:
-            logger.error(f"Error getting stream or profile: {e}")
+
+            # Get the M3U account profile for URL pattern
+            m3u_profile = profile
+
+            # Get the appropriate user agent
+            m3u_account = M3UAccount.objects.get(id=m3u_profile.m3u_account.id)
+            stream_user_agent = m3u_account.get_user_agent().user_agent
+
+            if stream_user_agent is None:
+                stream_user_agent = UserAgent.objects.get(id=CoreSettings.get_default_user_agent_id())
+                logger.debug(f"No user agent found for account, using default: {stream_user_agent}")
+
+            # Generate stream URL based on the selected profile
+            input_url = stream.url
+            stream_url = transform_url(input_url, m3u_profile.search_pattern, m3u_profile.replace_pattern)
+
+            # Check if transcoding is needed
+            stream_profile = channel.get_stream_profile()
+            if stream_profile.is_proxy() or stream_profile is None:
+                transcode = False
+            else:
+                transcode = True
+
+            stream_profile_id = stream_profile.id
+
+            return stream_url, stream_user_agent, transcode, stream_profile_id
+        except Exception as e:
+            logger.error(f"Error generating stream URL for channel {channel_id}: {e}")
+            if not channel.release_stream():
+                logger.warning(f"Failed to release stream for channel {channel_id} after URL generation error")
             return None, None, False, None
-
-        # Get the M3U account profile for URL pattern
-        m3u_profile = profile
-
-        # Get the appropriate user agent
-        m3u_account = M3UAccount.objects.get(id=m3u_profile.m3u_account.id)
-        stream_user_agent = m3u_account.get_user_agent().user_agent
-
-        if stream_user_agent is None:
-            stream_user_agent = UserAgent.objects.get(id=CoreSettings.get_default_user_agent_id())
-            logger.debug(f"No user agent found for account, using default: {stream_user_agent}")
-
-        # Generate stream URL based on the selected profile
-        input_url = stream.url
-        stream_url = transform_url(input_url, m3u_profile.search_pattern, m3u_profile.replace_pattern)
-
-        # Check if transcoding is needed
-        stream_profile = channel.get_stream_profile()
-        if stream_profile.is_proxy() or stream_profile is None:
-            transcode = False
-        else:
-            transcode = True
-
-        stream_profile_id = stream_profile.id
-
-        return stream_url, stream_user_agent, transcode, stream_profile_id
     except Exception as e:
         logger.error(f"Error generating stream URL: {e}")
         return None, None, False, None
diff --git a/apps/proxy/ts_proxy/views.py b/apps/proxy/ts_proxy/views.py
index 91f254a7..ccbb947d 100644
--- a/apps/proxy/ts_proxy/views.py
+++ b/apps/proxy/ts_proxy/views.py
@@ -54,6 +54,7 @@ def stream_ts(request, channel_id):
 
     client_user_agent = None
     proxy_server = ProxyServer.get_instance()
+    connection_allocated = False  # Track if connection slot was allocated via get_stream()
 
     try:
         # Generate a unique client ID
@@ -219,9 +220,10 @@ def stream_ts(request, channel_id):
                     )
 
             if stream_url is None:
-                # Release the channel's stream lock if one was acquired
-                # Note: Only call this if get_stream() actually assigned a stream
-                # In our case, if stream_url is None, no stream was ever assigned, so don't release
+                # Release any connection slot that may have been allocated
+                # by the error-checking get_stream() call during retries
+                if not channel.release_stream():
+                    logger.debug(f"[{client_id}] release_stream found no keys during failed init cleanup")
 
                 # Get the specific error message if available
                 wait_duration = f"{int(time.time() - wait_start_time)}s"
@@ -237,8 +239,23 @@ def stream_ts(request, channel_id):
                     {"error": error_msg, "waited": wait_duration}, status=503
                 )  # 503 Service Unavailable is appropriate here
 
-            # Get the stream ID from the channel
-            stream_id, m3u_profile_id, _ = channel.get_stream()
+            # generate_stream_url() called get_stream() which allocated a connection
+            # slot (INCR'd profile_connections) - track this for cleanup on error
+            if needs_initialization:
+                connection_allocated = True
+
+            # Read stream assignment from Redis (already set by generate_stream_url → get_stream).
+            # Avoid calling get_stream() again — (INCR profile counter)
+            # It could double-allocate if the keys were cleared by a concurrent release.
+            stream_id = None
+            m3u_profile_id = None
+            if proxy_server.redis_client:
+                stream_id_bytes = proxy_server.redis_client.get(f"channel_stream:{channel.id}")
+                if stream_id_bytes:
+                    stream_id = int(stream_id_bytes)
+                    profile_id_bytes = proxy_server.redis_client.get(f"stream_profile:{stream_id}")
+                    if profile_id_bytes:
+                        m3u_profile_id = int(profile_id_bytes)
             logger.info(
                 f"Channel {channel_id} using stream ID {stream_id}, m3u account profile ID {m3u_profile_id}"
             )
@@ -308,7 +325,9 @@ def stream_ts(request, channel_id):
                                 f"[{client_id}] Alternate stream #{alt['stream_id']} failed validation: {message}"
                             )
                 # Release stream lock before redirecting
-                channel.release_stream()
+                if not channel.release_stream():
+                    logger.warning(f"[{client_id}] Failed to release stream before redirect")
+                connection_allocated = False
                 # Final decision based on validation results
                 if is_valid:
                     logger.info(
@@ -344,10 +363,17 @@ def stream_ts(request, channel_id):
             )
 
             if not success:
+                if connection_allocated:
+                    if not channel.release_stream():
+                        logger.warning(f"[{client_id}] Failed to release stream after init failure")
+                    connection_allocated = False
                 return JsonResponse(
                     {"error": "Failed to initialize channel"}, status=500
                 )
 
+            # Channel initialized - cleanup lifecycle now owns the connection release
+            connection_allocated = False
+
             # If we're the owner, wait for connection to establish
             if proxy_server.am_i_owner(channel_id):
                 manager = proxy_server.stream_managers.get(channel_id)
@@ -507,6 +533,12 @@ def stream_ts(request, channel_id):
 
     except Exception as e:
         logger.error(f"Error in stream_ts: {e}", exc_info=True)
+        if connection_allocated:
+            try:
+                if not channel.release_stream():
+                    logger.warning(f"[{client_id}] Failed to release stream in exception handler")
+            except Exception:
+                pass
         return JsonResponse({"error": str(e)}, status=500)
 
 
diff --git a/apps/proxy/vod_proxy/connection_manager.py b/apps/proxy/vod_proxy/connection_manager.py
index c4c2130f..7bfccfca 100644
--- a/apps/proxy/vod_proxy/connection_manager.py
+++ b/apps/proxy/vod_proxy/connection_manager.py
@@ -1402,17 +1402,25 @@ class VODConnectionManager:
                         client_id = session_id
 
                         # Remove individual connection tracking keys created during streaming
+                        # Check if connection key exists first - remove_connection()
+                        # handles the profile DECR when the key exists, but skips it
+                        # if the key already expired by TTL
+                        connection_key_exists = False
                         if content_type and content_uuid:
+                            connection_key = self._get_connection_key(content_type, content_uuid, client_id)
+                            connection_key_exists = bool(self.redis_client.exists(connection_key))
                             logger.info(f"[{session_id}] Cleaning up connection tracking keys")
                             self.remove_connection(content_type, content_uuid, client_id)
 
-                        # Remove from profile connections if counted (additional safety check)
-                        if session_data.get(b'connection_counted') == b'True' and profile_id:
-                            profile_key = self._get_profile_connections_key(int(profile_id.decode('utf-8')))
-                            current_count = int(self.redis_client.get(profile_key) or 0)
-                            if current_count > 0:
-                                self.redis_client.decr(profile_key)
-                                logger.info(f"[{session_id}] Decremented profile {profile_id.decode('utf-8')} connections")
+                        # Fallback DECR: only if the connection tracking key had already
+                        # expired (remove_connection skips DECR in that case)
+                        if not connection_key_exists:
+                            if session_data.get(b'connection_counted') == b'True' and profile_id:
+                                profile_key = self._get_profile_connections_key(int(profile_id.decode('utf-8')))
+                                current_count = int(self.redis_client.get(profile_key) or 0)
+                                if current_count > 0:
+                                    self.redis_client.decr(profile_key)
+                                    logger.info(f"[{session_id}] Decremented profile {profile_id.decode('utf-8')} connections (fallback)")
 
                     # Remove session tracking key
                     self.redis_client.delete(session_key)
diff --git a/apps/proxy/vod_proxy/multi_worker_connection_manager.py b/apps/proxy/vod_proxy/multi_worker_connection_manager.py
index 74c5db41..6b048529 100644
--- a/apps/proxy/vod_proxy/multi_worker_connection_manager.py
+++ b/apps/proxy/vod_proxy/multi_worker_connection_manager.py
@@ -1154,9 +1154,10 @@ class MultiWorkerVODConnectionManager:
                 self._decrement_profile_connections(m3u_profile.id)
 
                 # Also clean up the Redis connection state since we won't be using it
+                # Pass connection_manager=None since we already decremented above
                 if redis_connection:
                     try:
-                        redis_connection.cleanup(connection_manager=self, current_worker_id=self.worker_id)
+                        redis_connection.cleanup(current_worker_id=self.worker_id)
                     except Exception as cleanup_error:
                         logger.error(f"[{client_id}] Error during cleanup after connection failure: {cleanup_error}")
 
@@ -1380,12 +1381,14 @@ class MultiWorkerVODConnectionManager:
                 profile_id = connection_data.get('m3u_profile_id')
                 if profile_id:
                     profile_connections_key = f"profile_connections:{profile_id}"
+                    current_count = int(self.redis_client.get(profile_connections_key) or 0)
 
                     # Use pipeline for atomic operations
                     pipe = self.redis_client.pipeline()
                     pipe.delete(connection_key)
                     pipe.srem(content_connections_key, client_id)
-                    pipe.decr(profile_connections_key)
+                    if current_count > 0:
+                        pipe.decr(profile_connections_key)
                     pipe.execute()
 
                     logger.info(f"Removed Redis-backed connection: {client_id}")
diff --git a/core/api_views.py b/core/api_views.py
index 3d131ab9..43e55d02 100644
--- a/core/api_views.py
+++ b/core/api_views.py
@@ -183,6 +183,7 @@ class ProxySettingsViewSet(viewsets.ViewSet):
                 "redis_chunk_ttl": 60,
                 "channel_shutdown_delay": 0,
                 "channel_init_grace_period": 5,
+                "new_client_behind_seconds": 2,
             }
             settings_obj, created = CoreSettings.objects.get_or_create(
                 key=PROXY_SETTINGS_KEY,
diff --git a/core/models.py b/core/models.py
index 10f9073f..7ce7713d 100644
--- a/core/models.py
+++ b/core/models.py
@@ -329,6 +329,7 @@ class CoreSettings(models.Model):
             "redis_chunk_ttl": 60,
             "channel_shutdown_delay": 0,
             "channel_init_grace_period": 5,
+            "new_client_behind_seconds": 2,
         })
 
     # System Settings
diff --git a/core/serializers.py b/core/serializers.py
index 03ba33e8..b0b730ee 100644
--- a/core/serializers.py
+++ b/core/serializers.py
@@ -78,6 +78,7 @@ class ProxySettingsSerializer(serializers.Serializer):
     redis_chunk_ttl = serializers.IntegerField(min_value=10, max_value=3600)
     channel_shutdown_delay = serializers.IntegerField(min_value=0, max_value=300)
     channel_init_grace_period = serializers.IntegerField(min_value=0, max_value=60)
+    new_client_behind_seconds = serializers.IntegerField(min_value=0, max_value=120, required=False, default=2)
 
     def validate_buffering_timeout(self, value):
         if value < 0 or value > 300:
@@ -104,6 +105,11 @@ class ProxySettingsSerializer(serializers.Serializer):
             raise serializers.ValidationError("Channel init grace period must be between 0 and 60 seconds")
         return value
 
+    def validate_new_client_behind_seconds(self, value):
+        if value < 0 or value > 120:
+            raise serializers.ValidationError("New client buffer must be between 0 and 120 seconds")
+        return value
+
 
 class SystemNotificationSerializer(serializers.ModelSerializer):
     """Serializer for system notifications."""
diff --git a/frontend/src/components/forms/settings/ProxySettingsForm.jsx b/frontend/src/components/forms/settings/ProxySettingsForm.jsx
index a5c8d858..52440769 100644
--- a/frontend/src/components/forms/settings/ProxySettingsForm.jsx
+++ b/frontend/src/components/forms/settings/ProxySettingsForm.jsx
@@ -24,6 +24,7 @@ const ProxySettingsOptions = React.memo(({ proxySettingsForm }) => {
       'redis_chunk_ttl',
       'channel_shutdown_delay',
       'channel_init_grace_period',
+      'new_client_behind_seconds',
     ].includes(key);
   };
   const isFloatField = (key) => {
@@ -36,7 +37,9 @@ const ProxySettingsOptions = React.memo(({ proxySettingsForm }) => {
         ? 3600
         : key === 'channel_shutdown_delay'
           ? 300
-          : 60;
+          : key === 'new_client_behind_seconds'
+            ? 120
+            : 60;
   };
   return (
     <>
@@ -97,7 +100,12 @@ const ProxySettingsForm = React.memo(({ active }) => {
   useEffect(() => {
     if (settings) {
       if (settings['proxy_settings']?.value) {
-        proxySettingsForm.setValues(settings['proxy_settings'].value);
+        // Merge defaults so any newly-added keys not yet in the stored
+        // settings object still show their default value rather than blank.
+        proxySettingsForm.setValues({
+          ...getProxySettingDefaults(),
+          ...settings['proxy_settings'].value,
+        });
       }
     }
   }, [settings]);
diff --git a/frontend/src/constants.js b/frontend/src/constants.js
index d2caa45b..0c19eea1 100644
--- a/frontend/src/constants.js
+++ b/frontend/src/constants.js
@@ -55,6 +55,11 @@ export const PROXY_SETTINGS_OPTIONS = {
     label: 'Channel Initialization Grace Period',
     description: 'Grace period in seconds during channel initialization',
   },
+  new_client_behind_seconds: {
+    label: 'New Client Buffer (seconds)',
+    description:
+      'Seconds of received buffer to start behind live when a new client connects (0 = start at live). Note: this is chunk receive time, not video duration.',
+  },
 };
 
 export const M3U_FILTER_TYPES = [
diff --git a/frontend/src/utils/forms/settings/ProxySettingsFormUtils.js b/frontend/src/utils/forms/settings/ProxySettingsFormUtils.js
index 9ca185e8..e6e0464b 100644
--- a/frontend/src/utils/forms/settings/ProxySettingsFormUtils.js
+++ b/frontend/src/utils/forms/settings/ProxySettingsFormUtils.js
@@ -14,5 +14,6 @@ export const getProxySettingDefaults = () => {
     redis_chunk_ttl: 60,
     channel_shutdown_delay: 0,
     channel_init_grace_period: 5,
+    new_client_behind_seconds: 2,
   };
 };
diff --git a/frontend/src/utils/forms/settings/__tests__/ProxySettingsFormUtils.test.js b/frontend/src/utils/forms/settings/__tests__/ProxySettingsFormUtils.test.js
index 8feefccb..a48e5a37 100644
--- a/frontend/src/utils/forms/settings/__tests__/ProxySettingsFormUtils.test.js
+++ b/frontend/src/utils/forms/settings/__tests__/ProxySettingsFormUtils.test.js
@@ -61,6 +61,7 @@ describe('ProxySettingsFormUtils', () => {
         redis_chunk_ttl: 60,
         channel_shutdown_delay: 0,
         channel_init_grace_period: 5,
+        new_client_behind_seconds: 2,
       });
     });
 
@@ -80,6 +81,7 @@ describe('ProxySettingsFormUtils', () => {
       expect(typeof result.redis_chunk_ttl).toBe('number');
       expect(typeof result.channel_shutdown_delay).toBe('number');
       expect(typeof result.channel_init_grace_period).toBe('number');
+      expect(typeof result.new_client_behind_seconds).toBe('number');
     });
   });
 });