eminux/MatchLiveTv
1
1
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

Rende sticky e capacizzati i relay YouTube su coda dedicata.

Browse Source 
Evita doppi ffmpeg tra host: stop solo sull'owner, ensure con requeue a capacità piena e coda Sidekiq youtube_relay isolata.

Co-authored-by: Cursor <cursoragent@cursor.com>
This commit is contained in:
eminux
2026-08-09 19:54:04 +02:00
co-authored by Cursor
parent c3878cdc6d
commit 0b55d5a8fa
9 changed files with 171 additions and 29 deletions
Download Patch File Download Diff File Expand all files Collapse all files
backend/app/services/streams/youtube_relay.rb
+80 -20
View File
@@ -1,29 +1,35 @@
module Streams
# Relay verso YouTube: legge RTMP/HLS da MediaMTX e inoltra su RTMPS (-c copy). Nessun overlay.
# ffmpeg gira solo nel container sidekiq (YOUTUBE_RELAY_WORKER=1).
# ffmpeg gira solo sui worker Sidekiq con YOUTUBE_RELAY_WORKER=1 (coda youtube_relay).
class YoutubeRelay
class Error < StandardError; end
REDIS_KEY = "youtube_relay:pid:%s"
OWNER_KEY = "youtube_relay:owner:%s"
OWNED_SET = "youtube_relay:owned:%s"
QUEUE = :youtube_relay
class << self
def worker?
ENV["YOUTUBE_RELAY_WORKER"] == "1"
end
def max_concurrent
ENV.fetch("RELAY_MAX_CONCURRENT", "4").to_i
end
def start(session)
return unless worker?
start_on_worker!(session)
end
# Non cancella owner/pid qui: solo il worker owner deve killare ffmpeg.
def stop(session)
clear_pid(session.id)
redis.del(format(OWNER_KEY, session.id))
if worker?
if worker? && owner_is_local?(session.id)
stop_on_worker!(session)
else
YoutubeRelayStopJob.perform_later(session.id)
YoutubeRelayStopJob.set(queue: QUEUE).perform_later(session.id)
end
true
end
@@ -31,11 +37,13 @@ module Streams
def running?(session_id)
owner = redis.get(format(OWNER_KEY, session_id))
pid = pid_for(session_id)
return false if pid.blank? && owner.blank?
return false if pid.blank? && owner.present? && owner != worker_id && redis.ttl(format(OWNER_KEY, session_id)) <= 30
return false if pid.blank?
return process_alive?(pid) if owner.blank? || owner == worker_id
# Relay avviato in un altro container: consideralo attivo se il lock è recente.
# Relay su altro host: attivo se lock owner ancora fresco.
redis.ttl(format(OWNER_KEY, session_id)) > 30
end
@@ -47,12 +55,12 @@ module Streams
if worker?
ensure_on_worker!(session)
else
YoutubeRelayEnsureJob.perform_later(session.id)
YoutubeRelayEnsureJob.set(queue: QUEUE).perform_later(session.id)
end
end
def ensure_on_worker!(session)
return unless worker?
return :not_worker unless worker?
return unless session.platform == "youtube"
return if session.terminal?
return if session.stream_key.blank?
@@ -60,28 +68,64 @@ module Streams
return unless intake_available?(session)
pid = pid_for(session.id)
clear_pid(session.id) if pid.present? && !process_alive?(pid.to_i)
if pid.present? && !process_alive?(pid.to_i)
clear_local_ownership(session.id)
end
return if running?(session.id)
if running?(session.id)
touch_owner!(session.id) if owner_is_local?(session.id)
return :already_running
end
if at_capacity?
YoutubeRelayEnsureJob.set(wait: 5.seconds, queue: QUEUE).perform_later(session.id)
Rails.logger.info("[YoutubeRelay] at capacity worker=#{worker_id} session=#{session.id} requeue")
return :at_capacity
end
last_restart = redis.get(restart_debounce_key(session.id)).to_i
return if last_restart.positive? && (Time.now.to_i - last_restart) < 5
return :debounced if last_restart.positive? && (Time.now.to_i - last_restart) < 5
start_on_worker!(session)
redis.set(restart_debounce_key(session.id), Time.now.to_i, ex: 300)
:started
rescue Error => e
Rails.logger.warn("[YoutubeRelay] ensure_on_worker session=#{session.id}: #{e.message}")
:error
end
# @return [Symbol] :stopped, :wrong_host, :noop
def stop_on_worker!(session)
return :not_worker unless worker?
owner = redis.get(format(OWNER_KEY, session.id))
if owner.present? && owner != worker_id
return :wrong_host
end
pid = pid_for(session.id)
return false if pid.blank?
if pid.blank?
clear_local_ownership(session.id)
return :noop
end
terminate_pid(pid)
clear_pid(session.id)
redis.del(format(OWNER_KEY, session.id))
Rails.logger.info("[YoutubeRelay] stopped pid=#{pid} session=#{session.id}")
true
clear_local_ownership(session.id)
Rails.logger.info("[YoutubeRelay] stopped pid=#{pid} session=#{session.id} worker=#{worker_id}")
:stopped
end
def local_owned_count
redis.scard(format(OWNED_SET, worker_id)).to_i
end
def at_capacity?
local_owned_count >= max_concurrent
end
def owner_is_local?(session_id)
owner = redis.get(format(OWNER_KEY, session_id))
owner.blank? || owner == worker_id
end
private
@@ -92,9 +136,9 @@ module Streams
return if session.terminal?
return unless intake_available?(session)
return pid_for(session.id).to_i if running?(session.id)
return pid_for(session.id).to_i if running?(session.id) && owner_is_local?(session.id)
stop_on_worker!(session) if pid_for(session.id).present?
stop_on_worker!(session) if pid_for(session.id).present? && owner_is_local?(session.id)
log_path = log_file(session)
FileUtils.mkdir_p(File.dirname(log_path))
@@ -108,8 +152,8 @@ module Streams
)
Process.detach(pid)
store_pid(session.id, pid)
redis.set(format(OWNER_KEY, session.id), worker_id, ex: 48.hours.to_i)
Rails.logger.info("[YoutubeRelay] started pid=#{pid} session=#{session.id} intake=#{intake_source.join(":")}")
claim_ownership!(session.id)
Rails.logger.info("[YoutubeRelay] started pid=#{pid} session=#{session.id} intake=#{intake_source.join(":")} worker=#{worker_id}")
schedule_youtube_activate(session)
pid
rescue Errno::ENOENT => e
@@ -185,6 +229,22 @@ module Streams
format("youtube_relay:debounce:%s", session_id)
end
def claim_ownership!(session_id)
redis.set(format(OWNER_KEY, session_id), worker_id, ex: 48.hours.to_i)
redis.sadd(format(OWNED_SET, worker_id), session_id)
end
def touch_owner!(session_id)
redis.expire(format(OWNER_KEY, session_id), 48.hours.to_i)
redis.expire(format(REDIS_KEY, session_id), 48.hours.to_i)
end
def clear_local_ownership(session_id)
redis.del(format(REDIS_KEY, session_id))
redis.del(format(OWNER_KEY, session_id))
redis.srem(format(OWNED_SET, worker_id), session_id)
end
def store_pid(session_id, pid)
redis.set(format(REDIS_KEY, session_id), pid, ex: 48.hours.to_i)
end