diff --git a/backend/app/jobs/youtube_relay_ensure_job.rb b/backend/app/jobs/youtube_relay_ensure_job.rb index 3761657..bf75674 100644 --- a/backend/app/jobs/youtube_relay_ensure_job.rb +++ b/backend/app/jobs/youtube_relay_ensure_job.rb @@ -1,6 +1,6 @@ -# Avvia/riavvia il relay YouTube solo nel container sidekiq (YOUTUBE_RELAY_WORKER=1). +# Avvia/riavvia il relay YouTube solo sui worker con YOUTUBE_RELAY_WORKER=1 (coda youtube_relay). class YoutubeRelayEnsureJob < ApplicationJob - queue_as :default + queue_as Streams::YoutubeRelay::QUEUE def perform(session_id) session = StreamSession.find_by(id: session_id) diff --git a/backend/app/jobs/youtube_relay_stop_job.rb b/backend/app/jobs/youtube_relay_stop_job.rb index 202559d..cf868f3 100644 --- a/backend/app/jobs/youtube_relay_stop_job.rb +++ b/backend/app/jobs/youtube_relay_stop_job.rb @@ -1,10 +1,22 @@ +# Ferma il relay solo sull'owner. Se il job gira su un altro host, requeue breve. class YoutubeRelayStopJob < ApplicationJob - queue_as :default + queue_as Streams::YoutubeRelay::QUEUE - def perform(session_id) + discard_on ActiveJob::DeserializationError + + def perform(session_id, attempts = 0) session = StreamSession.find_by(id: session_id) return unless session - Streams::YoutubeRelay.stop_on_worker!(session) if Streams::YoutubeRelay.worker? + unless Streams::YoutubeRelay.worker? + # Solo i worker relay processano questa coda in modo utile. + return + end + + result = Streams::YoutubeRelay.stop_on_worker!(session) + return unless result == :wrong_host + return if attempts >= 30 + + self.class.set(wait: 2.seconds).perform_later(session_id, attempts + 1) end end diff --git a/backend/app/services/mediamtx/publisher_sync.rb b/backend/app/services/mediamtx/publisher_sync.rb index fcc5c43..2f15ef9 100644 --- a/backend/app/services/mediamtx/publisher_sync.rb +++ b/backend/app/services/mediamtx/publisher_sync.rb @@ -183,7 +183,7 @@ module Mediamtx key = format("youtube_relay:sched:%s", session.id) return unless redis.set(key, "1", nx: true, ex: 10) - YoutubeRelayEnsureJob.perform_later(session.id) + YoutubeRelayEnsureJob.set(queue: Streams::YoutubeRelay::QUEUE).perform_later(session.id) end def redis diff --git a/backend/app/services/streams/youtube_relay.rb b/backend/app/services/streams/youtube_relay.rb index a1cfec4..eb9677d 100644 --- a/backend/app/services/streams/youtube_relay.rb +++ b/backend/app/services/streams/youtube_relay.rb @@ -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 diff --git a/backend/config/sidekiq.yml b/backend/config/sidekiq.yml index 3d67b00..bcaa07c 100644 --- a/backend/config/sidekiq.yml +++ b/backend/config/sidekiq.yml @@ -1,4 +1,5 @@ :concurrency: 5 :queues: - - default - critical + - youtube_relay + - default diff --git a/backend/spec/jobs/youtube_relay_stop_job_spec.rb b/backend/spec/jobs/youtube_relay_stop_job_spec.rb new file mode 100644 index 0000000..493feb8 --- /dev/null +++ b/backend/spec/jobs/youtube_relay_stop_job_spec.rb @@ -0,0 +1,18 @@ +# frozen_string_literal: true + +require "rails_helper" + +RSpec.describe YoutubeRelayStopJob, type: :job do + it "requeues when stop runs on the wrong host" do + session = instance_double(StreamSession, id: SecureRandom.uuid) + allow(StreamSession).to receive(:find_by).and_return(session) + allow(Streams::YoutubeRelay).to receive(:worker?).and_return(true) + allow(Streams::YoutubeRelay).to receive(:stop_on_worker!).and_return(:wrong_host) + + job_proxy = double("ConfiguredJob") + expect(described_class).to receive(:set).with(wait: 2.seconds).and_return(job_proxy) + expect(job_proxy).to receive(:perform_later).with(session.id, 1) + + described_class.new.perform(session.id, 0) + end +end diff --git a/backend/spec/services/streams/youtube_relay_spec.rb b/backend/spec/services/streams/youtube_relay_spec.rb index df31b96..08cd7be 100644 --- a/backend/spec/services/streams/youtube_relay_spec.rb +++ b/backend/spec/services/streams/youtube_relay_spec.rb @@ -21,4 +21,46 @@ RSpec.describe Streams::YoutubeRelay do expect(cmd).not_to include("-b:a") end end + + describe "multi-host sticky stop/capacity" do + let(:redis) { Redis.new(url: ENV.fetch("REDIS_URL", "redis://redis:6379/0")) } + let(:session_id) { SecureRandom.uuid } + + before do + allow(described_class).to receive(:worker?).and_return(true) + allow(described_class).to receive(:worker_id).and_return("worker-a") + allow(described_class).to receive(:max_concurrent).and_return(1) + redis.flushdb + end + + def session_double + instance_double( + StreamSession, + id: session_id, + platform: "youtube", + terminal?: false, + stream_key: "yt-key", + status: "live" + ) + end + + it "does not stop ffmpeg owned by another worker" do + redis.set(format(Streams::YoutubeRelay::OWNER_KEY, session_id), "worker-b", ex: 3600) + redis.set(format(Streams::YoutubeRelay::REDIS_KEY, session_id), "12345", ex: 3600) + + expect(described_class.stop_on_worker!(session_double)).to eq(:wrong_host) + expect(redis.get(format(Streams::YoutubeRelay::OWNER_KEY, session_id))).to eq("worker-b") + end + + it "requeues ensure when at capacity" do + other_id = SecureRandom.uuid + redis.sadd(format(Streams::YoutubeRelay::OWNED_SET, "worker-a"), other_id) + allow(described_class).to receive(:intake_available?).and_return(true) + expect(YoutubeRelayEnsureJob).to receive(:set).with(hash_including(wait: 5.seconds, queue: :youtube_relay)).and_return( + double(perform_later: true) + ) + + expect(described_class.ensure_on_worker!(session_double)).to eq(:at_capacity) + end + end end diff --git a/docs/infrastructure/STREAMING_AUTOSCALE.md b/docs/infrastructure/STREAMING_AUTOSCALE.md index dabaa5d..37badc9 100644 --- a/docs/infrastructure/STREAMING_AUTOSCALE.md +++ b/docs/infrastructure/STREAMING_AUTOSCALE.md @@ -209,7 +209,7 @@ Cosa il lab **non** replica al 100%: tempi boot Hetzner, vSwitch, RTMP 4G multi- | **L — Lab Proxmox** | `LocalLab` / `ProxmoxLab`, DNS lab, admin nodi | **implementata sul branch** | | **0 — Multi-nodo ready** | Registry + URL RTMP/HLS in API (`StreamNode`, `Streams::NodeRegistry`) | **implementata sul branch** | | **1 — Hetzner Cloud + DNS** | `HetznerCloudProvider` + `HetznerDnsProvider`, `mltv-stream.net`, cloud-init, WireGuard (§16) | **implementata sul branch** (WG ops manuale) | -| **2 — Routing relay** | Coda `youtube_relay`, sticky owner, cap | Niente doppi ffmpeg | +| **2 — Routing relay** | Coda `youtube_relay`, sticky owner, cap `RELAY_MAX_CONCURRENT` | **implementata sul branch** | | **3 — Autoscaler** | Soglie + warm spare (+1) | Automatico | | **4 — Hardening** | Drain, budget, runbook, kill-switch | Picchi weekend | | **A — Auction (futuro)** | Migrazione control plane + vSwitch al posto di WireGuard | Hardware dedicato Hetzner | @@ -242,6 +242,14 @@ Cosa il lab **non** replica al 100%: tempi boot Hetzner, vSwitch, RTMP 4G multi- - Decommission usa il provider del nodo (non solo ENV globale) - WireGuard: vedi §16 +### Fase 2 — dettagli implementati + +- Coda Sidekiq dedicata `youtube_relay` (priorità sopra `default`) +- `YoutubeRelayEnsureJob` / `YoutubeRelayStopJob` solo su quella coda +- Stop sticky: non cancella owner da Rails; lo stop gira sull’owner o requeue (`:wrong_host`) +- Cap per worker: `RELAY_MAX_CONCURRENT` (default 4) + set Redis `youtube_relay:owned:HOSTNAME` +- Ensure a capacità piena → requeue 5s invece di avviare un secondo ffmpeg locale + Ordine di lavoro consigliato sul branch: **0 → L → 1 → 2 → 3 → 4**; **A** quando si decide di lasciare il Proxmox casa. --- diff --git a/infra/.env.production.example b/infra/.env.production.example index 95b9d6d..1198f0e 100644 --- a/infra/.env.production.example +++ b/infra/.env.production.example @@ -121,4 +121,5 @@ STREAM_CLOUD_MAX_PUBLISHERS=4 STREAM_NODE_ENV=prod # Lab locale di default; il bottone admin "Provisiona nodo Hetzner" usa comunque i provider hetzner. STREAM_CLOUD_PROVIDER=local_lab -STREAM_DNS_PROVIDER=lab +RELAY_MAX_CONCURRENT=4 +YOUTUBE_RELAY_WORKER=1