# frozen_string_literal: true require "rails_helper" RSpec.describe Streams::Autoscaler do def with_env(vars) previous = vars.keys.index_with { |k| ENV[k] } vars.each { |k, v| ENV[k] = v } yield ensure previous.each { |k, v| v.nil? ? ENV.delete(k) : ENV[k] = v } end let(:redis) { Redis.new(url: ENV.fetch("REDIS_URL", "redis://redis:6379/0")) } before do redis.del(Streams::Autoscaler::LOCK_KEY) redis.del(Streams::Autoscaler::KILL_SWITCH_KEY) redis.del(Streams::DnsProviders::Lab::REDIS_KEY) StreamSession.where.not(stream_node_id: nil).update_all(stream_node_id: nil, status: "ended", ended_at: Time.current) StreamNode.where.not(slug: Streams::NodeRegistry::HOME_SLUG).delete_all ENV["STREAM_AUTOSCALE_QUIET_HOURS"] = "off" end after do ENV.delete("STREAM_AUTOSCALE_QUIET_HOURS") end it "is a no-op when disabled" do with_env("STREAM_AUTOSCALE_ENABLED" => "0") do result = described_class.reconcile! expect(result.skipped).to eq(true) expect(result.actions).to eq([]) end end it "scales out and creates a warm spare when free slots are low" do with_env( "STREAM_AUTOSCALE_ENABLED" => "1", "STREAM_AUTOSCALE_SOFT_FREE_SLOTS" => "2", "STREAM_AUTOSCALE_WARM_SPARE" => "1", "STREAM_AUTOSCALE_KIND" => "lab", "STREAM_AUTOSCALE_MAX_NODES" => "5", "STREAM_CLOUD_PROVIDER" => "local_lab", "STREAM_DNS_PROVIDER" => "lab", "STREAM_NODE_HOME_MAX_PUBLISHERS" => "2", "MEDIAMTX_API_URL" => "http://mtx-home:9997", "MEDIAMTX_RTMP_URL" => "rtmp://home.example:1935", "HLS_PUBLIC_URL" => "https://home.example/hls" ) do home = Streams::NodeRegistry.ensure_home_from_env! user = User.create!(email: "as@example.com", name: "A", password: "Password123", role: "coach") club = Club.create!(name: "AS", sport: "volleyball") team = club.teams.create!(name: "T", sport: "volleyball", slug: "as-t") match = team.matches.create!(opponent_name: "X", scheduled_at: 1.hour.from_now) # Fill home (2/2) → free_slots=0 2.times do StreamSession.create!(match: match, user: user, platform: "matchlivetv", status: "live", stream_node: home) end provisioner = instance_double(Streams::NodeProvisioner) created = [] allow(provisioner).to receive(:provision_lab!) do node = StreamNode.create!( slug: "ingest-lab-#{created.size + 1}", hostname: "h#{created.size}.lab", role: "lab", status: "ready", provider: "local", rtmp_base_url: "rtmp://h:1935", hls_base_url: "https://h/hls", api_base_url: "http://h:9997", max_publishers: 2, max_relays: 2 ) created << node node end result = described_class.reconcile!(provisioner: provisioner) expect(result.actions).to include(:scale_out) expect(created.size).to eq(1) expect(result.metrics[:free_slots]).to be >= 2 # Riempi il nodo overflow → spare_ready=0 ma carico presente → warm spare StreamSession.create!( match: match, user: user, platform: "matchlivetv", status: "live", stream_node: created.first ) StreamSession.create!( match: match, user: user, platform: "matchlivetv", status: "live", stream_node: created.first ) result2 = described_class.reconcile!(provisioner: provisioner) expect(result2.actions).to include(:warm_spare).or include(:scale_out) expect(created.size).to be >= 2 end end it "does not scale in a node provisioned in the same reconcile round" do with_env( "STREAM_AUTOSCALE_ENABLED" => "1", "STREAM_AUTOSCALE_SOFT_FREE_SLOTS" => "2", "STREAM_AUTOSCALE_WARM_SPARE" => "0", "STREAM_AUTOSCALE_IDLE_MINUTES" => "0", "STREAM_AUTOSCALE_KIND" => "lab", "STREAM_AUTOSCALE_MAX_NODES" => "5", "STREAM_CLOUD_PROVIDER" => "local_lab", "STREAM_DNS_PROVIDER" => "lab", "STREAM_NODE_HOME_MAX_PUBLISHERS" => "2", "MEDIAMTX_API_URL" => "http://mtx-home:9997", "MEDIAMTX_RTMP_URL" => "rtmp://home.example:1935", "HLS_PUBLIC_URL" => "https://home.example/hls" ) do home = Streams::NodeRegistry.ensure_home_from_env! user = User.create!(email: "same@example.com", name: "S", password: "Password123", role: "coach") club = Club.create!(name: "Same", sport: "volleyball") team = club.teams.create!(name: "T", sport: "volleyball", slug: "same-t") match = team.matches.create!(opponent_name: "X", scheduled_at: 1.hour.from_now) 2.times do StreamSession.create!(match: match, user: user, platform: "matchlivetv", status: "live", stream_node: home) end provisioner = instance_double(Streams::NodeProvisioner) created = nil allow(provisioner).to receive(:provision_lab!) do created = StreamNode.create!( slug: "ingest-lab-new", hostname: "new.lab", role: "lab", status: "ready", provider: "local", rtmp_base_url: "rtmp://h:1935", hls_base_url: "https://h/hls", api_base_url: "http://h:9997", max_publishers: 4, max_relays: 2 ) end expect(provisioner).not_to receive(:decommission!) result = described_class.reconcile!(provisioner: provisioner) expect(result.actions).to include(:scale_out) expect(result.actions.map(&:to_s)).not_to include("scale_in_ingest-lab-new") expect(StreamNode.find_by(slug: "ingest-lab-new")).to be_present end end it "scales in an idle overflow node when warm spare is not required" do with_env( "STREAM_AUTOSCALE_ENABLED" => "1", "STREAM_AUTOSCALE_SOFT_FREE_SLOTS" => "0", "STREAM_AUTOSCALE_WARM_SPARE" => "0", "STREAM_AUTOSCALE_IDLE_MINUTES" => "0", "STREAM_NODE_HOME_MAX_PUBLISHERS" => "6", "MEDIAMTX_API_URL" => "http://mtx-home:9997", "MEDIAMTX_RTMP_URL" => "rtmp://home.example:1935", "HLS_PUBLIC_URL" => "https://home.example/hls" ) do Streams::NodeRegistry.ensure_home_from_env! idle = StreamNode.create!( slug: "ingest-lab-99", hostname: "idle.lab", role: "lab", status: "ready", provider: "local", provider_instance_id: "sim-x", rtmp_base_url: "rtmp://h:1935", hls_base_url: "https://h/hls", api_base_url: "http://h:9997", max_publishers: 2, max_relays: 2, created_at: 2.hours.ago ) provisioner = instance_double(Streams::NodeProvisioner) allow(provisioner).to receive(:drain!) expect(provisioner).to receive(:decommission!).with(idle) result = described_class.reconcile!(provisioner: provisioner) expect(result.actions.map(&:to_s)).to include("scale_in_ingest-lab-99") end end it "skips when Redis kill-switch is engaged even if ENV is enabled" do redis.del(described_class::KILL_SWITCH_KEY) with_env("STREAM_AUTOSCALE_ENABLED" => "1") do described_class.engage_kill_switch! expect(described_class.enabled?).to eq(false) result = described_class.reconcile! expect(result.skipped).to eq(true) ensure described_class.clear_kill_switch! end end it "blocks scale-out when next node would exceed monthly budget" do with_env( "STREAM_AUTOSCALE_ENABLED" => "1", "STREAM_AUTOSCALE_SOFT_FREE_SLOTS" => "2", "STREAM_AUTOSCALE_WARM_SPARE" => "0", "STREAM_AUTOSCALE_KIND" => "lab", "STREAM_AUTOSCALE_MAX_NODES" => "5", "STREAM_AUTOSCALE_NODE_EUR_PER_HOUR" => "1", "STREAM_AUTOSCALE_MONTHLY_BUDGET_EUR" => "10", "STREAM_CLOUD_PROVIDER" => "local_lab", "STREAM_DNS_PROVIDER" => "lab", "STREAM_NODE_HOME_MAX_PUBLISHERS" => "1", "MEDIAMTX_API_URL" => "http://mtx-home:9997", "MEDIAMTX_RTMP_URL" => "rtmp://home.example:1935", "HLS_PUBLIC_URL" => "https://home.example/hls" ) do StreamSession.where.not(stream_node_id: nil).update_all(stream_node_id: nil, status: "ended", ended_at: Time.current) StreamNode.where.not(slug: Streams::NodeRegistry::HOME_SLUG).delete_all home = Streams::NodeRegistry.ensure_home_from_env! home.update!(max_publishers: 1) user = User.create!(email: "budget@example.com", name: "B", password: "Password123", role: "coach") club = Club.create!(name: "Budget", sport: "volleyball") team = club.teams.create!(name: "T", sport: "volleyball", slug: "budget-t") match = team.matches.create!(opponent_name: "X", scheduled_at: 1.hour.from_now) StreamSession.create!(match: match, user: user, platform: "matchlivetv", status: "live", stream_node: home) provisioner = instance_double(Streams::NodeProvisioner) expect(provisioner).not_to receive(:provision_lab!) result = described_class.reconcile!(provisioner: provisioner) expect(result.actions).to include(:blocked_capacity) # 0 overflow nodes → stima 0€ ≤ budget; il blocco è sul nodo successivo (+1) expect(described_class.within_budget?(0)).to eq(true) expect(described_class.within_budget?(1)).to eq(false) end end it "promotes provisioning nodes when MediaMTX becomes reachable" do with_env( "STREAM_AUTOSCALE_ENABLED" => "1", "STREAM_AUTOSCALE_WARM_SPARE" => "0", "STREAM_AUTOSCALE_SOFT_FREE_SLOTS" => "0", "STREAM_AUTOSCALE_KIND" => "lab", "MEDIAMTX_API_URL" => "http://mtx-home:9997", "MEDIAMTX_RTMP_URL" => "rtmp://home.example:1935", "HLS_PUBLIC_URL" => "https://home.example/hls" ) do Streams::NodeRegistry.ensure_home_from_env! node = StreamNode.create!( slug: "ingest-lab-prov", hostname: "prov.lab", role: "lab", status: "provisioning", provider: "local", rtmp_base_url: "rtmp://h:1935", hls_base_url: "https://h/hls", api_base_url: "http://h:9997", max_publishers: 2, max_relays: 2 ) allow(Streams::NodeHealth).to receive(:promote_if_healthy!) do |n| n.update!(status: "ready", last_health_at: Time.current) true end provisioner = instance_double(Streams::NodeProvisioner) expect(provisioner).not_to receive(:provision_lab!) result = described_class.reconcile!(provisioner: provisioner) expect(result.actions).to include(:"ready_ingest-lab-prov") expect(node.reload.status).to eq("ready") end end it "blocks scale-out and warm spare during quiet hours" do with_env( "STREAM_AUTOSCALE_ENABLED" => "1", "STREAM_AUTOSCALE_QUIET_HOURS" => "02:00-07:00", "STREAM_AUTOSCALE_SOFT_FREE_SLOTS" => "2", "STREAM_AUTOSCALE_WARM_SPARE" => "1", "STREAM_AUTOSCALE_KIND" => "lab", "STREAM_AUTOSCALE_MAX_NODES" => "5", "STREAM_NODE_HOME_MAX_PUBLISHERS" => "1", "MEDIAMTX_API_URL" => "http://mtx-home:9997", "MEDIAMTX_RTMP_URL" => "rtmp://home.example:1935", "HLS_PUBLIC_URL" => "https://home.example/hls" ) do allow(Streams::QuietHours).to receive(:active?).and_return(true) home = Streams::NodeRegistry.ensure_home_from_env! user = User.create!(email: "qh@example.com", name: "Q", password: "Password123", role: "coach") club = Club.create!(name: "QH", sport: "volleyball") team = club.teams.create!(name: "T", sport: "volleyball", slug: "qh-t") match = team.matches.create!(opponent_name: "X", scheduled_at: 1.hour.from_now) StreamSession.create!(match: match, user: user, platform: "matchlivetv", status: "live", stream_node: home) provisioner = instance_double(Streams::NodeProvisioner) expect(provisioner).not_to receive(:provision_lab!) result = described_class.reconcile!(provisioner: provisioner) expect(result.actions).to include(:blocked_capacity) expect(result.metrics[:quiet_hours]).to eq(true) end end end