Fix race create sessione su CPX: ready solo dopo MediaMTX.

I nodi cloud restano in provisioning finché :9997 risponde; retry su create_path e 503 retryable se l’ingest è ancora irraggiungibile.

Co-authored-by: Cursor <cursoragent@cursor.com>

backend/app/services/streams/autoscaler.rb

@@ -136,6 +136,8 @@ module Streams
 
     def reconcile!
       actions = []
+      actions.concat(promote_provisioning_nodes!)
+      actions.concat(reclaim_stuck_provisioning!)
       m = self.class.metrics
 
       if need_capacity?(m) && can_provision?(m)
@@ -170,6 +172,30 @@ module Streams
 
     private
 
+    def promote_provisioning_nodes!
+      actions = []
+      StreamNode.where(status: "provisioning").find_each do |node|
+        next unless Streams::NodeHealth.promote_if_healthy!(node)
+
+        actions << :"ready_#{node.slug}"
+        Rails.logger.info("[Streams::Autoscaler] promoted #{node.slug} to ready")
+      end
+      actions
+    end
+
+    def reclaim_stuck_provisioning!
+      actions = []
+      stuck_after = ENV.fetch("STREAM_NODE_PROVISIONING_STUCK_MINUTES", "15").to_i.minutes.ago
+      StreamNode.where(status: "provisioning").where("created_at < ?", stuck_after).find_each do |node|
+        @provisioner.decommission!(node)
+        actions << :"reclaim_#{node.slug}"
+        Rails.logger.warn("[Streams::Autoscaler] decommissioned stuck provisioning #{node.slug}")
+      rescue NodeProvisioner::BusyError, NodeProvisioner::Error => e
+        Rails.logger.warn("[Streams::Autoscaler] reclaim #{node.slug}: #{e.message}")
+      end
+      actions
+    end
+
     def need_capacity?(m)
       m[:free_slots] <= self.class.soft_free_slots
     end

backend/app/services/streams/ingest_unavailable_error.rb

@@ -0,0 +1,13 @@
+# frozen_string_literal: true
+
+module Streams
+  # Ingest MediaMTX non raggiungibile (es. CPX ancora in bootstrap). API → 503 retryable.
+  class IngestUnavailableError < StandardError
+    attr_reader :code
+
+    def initialize(message = "Ingest temporaneamente non disponibile", code: "stream_ingest_unavailable")
+      super(message)
+      @code = code
+    end
+  end
+end

backend/app/services/streams/node_health.rb

@@ -0,0 +1,27 @@
+# frozen_string_literal: true
+
+module Streams
+  # Probe reachability of MediaMTX API on a stream node (:9997).
+  class NodeHealth
+    def self.mediamtx_up?(node, timeout: 2)
+      new(node, timeout: timeout).mediamtx_up?
+    end
+
+    def self.promote_if_healthy!(node, timeout: 2)
+      return false unless node.status == "provisioning"
+      return false unless mediamtx_up?(node, timeout: timeout)
+
+      node.update!(status: "ready", last_health_at: Time.current)
+      true
+    end
+
+    def initialize(node, timeout: 2)
+      @node = node
+      @timeout = timeout
+    end
+
+    def mediamtx_up?
+      Mediamtx::Client.new(base_url: @node.api_base_url).reachable?(timeout: @timeout)
+    end
+  end
+end

backend/app/services/streams/node_provisioner.rb

@@ -80,11 +80,15 @@ module Streams
       urls = urls_for_node(role: role, home: home, hostname: hostname, simulated: simulated,
                            private_ip: private_ip, public_ip: ip, use_node_hostname: use_node_hostname)
 
-      StreamNode.create!(
+      # Simulated/lab che riusa MediaMTX home: subito ready. Cloud reale: provisioning
+      # finché :9997 risponde (evita allocate → Faraday Connection refused → 500).
+      initial_status = simulated ? "ready" : "provisioning"
+
+      node = StreamNode.create!(
         slug: slug,
         hostname: hostname,
         role: role,
-        status: "ready",
+        status: initial_status,
         provider: provider_name_for(cloud, role: role),
         provider_instance_id: instance.id,
         rtmp_base_url: urls.fetch(:rtmp_base_url),
@@ -102,6 +106,34 @@ module Streams
           "cloud_raw" => instance.raw
         }
       )
+
+      wait_until_mediamtx_ready!(node) unless simulated
+      node.reload
+    end
+
+    def wait_until_mediamtx_ready!(node, timeout: nil, interval: nil)
+      timeout ||= ENV.fetch("STREAM_NODE_READY_TIMEOUT_SECS", "180").to_i
+      interval ||= ENV.fetch("STREAM_NODE_READY_POLL_SECS", "3").to_f
+      deadline = Process.clock_gettime(Process::CLOCK_MONOTONIC) + timeout
+
+      loop do
+        if Streams::NodeHealth.promote_if_healthy!(node)
+          Rails.logger.info("[Streams::NodeProvisioner] #{node.slug} MediaMTX ready")
+          return node
+        end
+
+        remaining = deadline - Process.clock_gettime(Process::CLOCK_MONOTONIC)
+        if remaining <= 0
+          Rails.logger.warn(
+            "[Streams::NodeProvisioner] #{node.slug} ancora provisioning dopo #{timeout}s " \
+            "(api=#{node.api_base_url}); autoscaler continuerà a promuovere"
+          )
+          return node
+        end
+
+        sleep([interval, remaining].min)
+        node.reload
+      end
     end
 
     def urls_for_node(role:, home:, hostname:, simulated:, private_ip:, public_ip: nil, use_node_hostname:)