Enhancement: 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)".

CHANGELOG.md

@@ -9,6 +9,7 @@ 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

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

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"""

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"""

apps/proxy/ts_proxy/stream_buffer.py

@@ -45,14 +45,21 @@ 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
@@ -101,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
@@ -284,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}")
 
@@ -420,6 +442,63 @@ class StreamBuffer:
             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"""

apps/proxy/ts_proxy/stream_generator.py

@@ -186,10 +186,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.buffer = buffer

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,

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

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."""

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]);

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 = [

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,
   };
 };