diff --git a/backend/app/jobs/recordings/cleanup_local_job.rb b/backend/app/jobs/recordings/cleanup_local_job.rb index 2fc2543..6e152c0 100644 --- a/backend/app/jobs/recordings/cleanup_local_job.rb +++ b/backend/app/jobs/recordings/cleanup_local_job.rb @@ -5,9 +5,11 @@ module Recordings def perform result = Recordings::CleanupLocal.new.call + orphan = Mediamtx::CleanupOrphanPaths.new.call Rails.logger.info( "[Recordings::CleanupLocalJob] removed=#{result.removed} " \ - "freed=#{result.freed_bytes} skipped=#{result.skipped}" + "freed=#{result.freed_bytes} skipped=#{result.skipped} " \ + "orphan_paths=#{orphan.removed}" ) end end diff --git a/backend/app/services/mediamtx/cleanup_orphan_paths.rb b/backend/app/services/mediamtx/cleanup_orphan_paths.rb new file mode 100644 index 0000000..ef41745 --- /dev/null +++ b/backend/app/services/mediamtx/cleanup_orphan_paths.rb @@ -0,0 +1,94 @@ +# frozen_string_literal: true + +module Mediamtx + # Rimuove path MediaMTX live/match_{uuid} lasciati attivi dopo fine diretta. + class CleanupOrphanPaths + Result = Struct.new(:removed, :skipped, keyword_init: true) + + SESSION_PATH = %r{\Alive/match_([0-9a-f-]{36})(?:_air)?\z}i + + ACTIVE_SESSION_STATUSES = %w[live paused connecting reconnecting].freeze + PROCESSING_GRACE = -> { ENV.fetch("RECORDINGS_LOCAL_PROCESSING_GRACE_HOURS", "6").to_i.hours } + + def initialize(client: Client.new, logger: Rails.logger) + @client = client + @logger = logger + @removed = 0 + @skipped = 0 + @deleted_session_ids = Set.new + end + + def call + online = @client.online_path_names + + @client.list_paths.each do |item| + name = item["name"].to_s + session_id = session_id_from_path(name) + unless session_id + @skipped += 1 + next + end + + if online.include?(name) || item["online"] + @skipped += 1 + next + end + + next unless removable?(session_id) + + remove_session_paths(session_id, name) + end + + Result.new(removed: @removed, skipped: @skipped) + end + + def removable?(session_id) + session = StreamSession.find_by(id: session_id) + return true if session.nil? + + return false if ACTIVE_SESSION_STATUSES.include?(session.status) + + recording = Recording.find_by(stream_session: session) + return true if recording.nil? + + return false if recording.status == "processing" && processing_within_grace?(session) + + true + end + + private + + def session_id_from_path(path_name) + match = path_name.match(SESSION_PATH) + match&.[](1) + end + + def processing_within_grace?(session) + ended_at = session.ended_at || session.updated_at + ended_at.nil? || ended_at >= PROCESSING_GRACE.call.ago + end + + def remove_session_paths(session_id, path_name) + return if @deleted_session_ids.include?(session_id) + + session = StreamSession.find_by(id: session_id) + if session + @client.delete_path(session) + else + delete_orphan_path_names(session_id) + end + + @deleted_session_ids.add(session_id) + @removed += 1 + @logger.info("[Mediamtx::CleanupOrphanPaths] removed paths for session #{session_id} (from #{path_name})") + rescue Client::Error, Faraday::Error => e + @skipped += 1 + @logger.warn("[Mediamtx::CleanupOrphanPaths] skip #{path_name}: #{e.message}") + end + + def delete_orphan_path_names(session_id) + @client.delete_path_name("live/match_#{session_id}") + @client.delete_path_name("live/match_#{session_id}_air") + end + end +end diff --git a/backend/app/services/sessions/stop.rb b/backend/app/services/sessions/stop.rb index e75357f..2217024 100644 --- a/backend/app/services/sessions/stop.rb +++ b/backend/app/services/sessions/stop.rb @@ -12,11 +12,8 @@ module Sessions complete_youtube_broadcast! if @session.youtube_broadcast_id.present? recording = Recordings::FinalizeSession.new(@session).call - if recording - Recordings::UploadJob.perform_async(@session.id) - else - Mediamtx::Client.new.delete_path(@session) - end + Recordings::UploadJob.perform_async(@session.id) if recording + remove_mediamtx_paths! log_event("ended") SessionChannel.broadcast_message(@session, { type: "stream_event", event: "ended" }) @@ -35,6 +32,12 @@ module Sessions @session.stream_events.create!(event_type: type, occurred_at: Time.current, metadata: {}) end + def remove_mediamtx_paths! + Mediamtx::Client.new.delete_path(@session) + rescue Mediamtx::Client::Error => e + Rails.logger.warn("[Sessions::Stop] delete_path #{@session.id}: #{e.message}") + end + def complete_youtube_broadcast! Youtube::BroadcastService.new(@session.match.team).complete_broadcast!(@session.youtube_broadcast_id) rescue Youtube::BroadcastService::Error => e diff --git a/backend/lib/tasks/recordings.rake b/backend/lib/tasks/recordings.rake index 40fbe80..ff07688 100644 --- a/backend/lib/tasks/recordings.rake +++ b/backend/lib/tasks/recordings.rake @@ -16,5 +16,8 @@ namespace :recordings do result = Recordings::CleanupLocal.new.call puts "Pulizia segmenti locali: rimossi=#{result.removed} " \ "liberati=#{result.freed_bytes} byte saltati=#{result.skipped}" + + orphan = Mediamtx::CleanupOrphanPaths.new.call + puts "Path MediaMTX orfani rimossi: #{orphan.removed} (saltati: #{orphan.skipped})" end end diff --git a/backend/lib/tasks/stream_sessions.rake b/backend/lib/tasks/stream_sessions.rake index 81eb644..57f4ff6 100644 --- a/backend/lib/tasks/stream_sessions.rake +++ b/backend/lib/tasks/stream_sessions.rake @@ -19,32 +19,14 @@ namespace :streams do puts " ended #{session.id} (fallback: #{e.message})" end - removed = cleanup_orphan_mediamtx_paths(client, online) - puts "Path MediaMTX rimossi: #{removed}" + result = Mediamtx::CleanupOrphanPaths.new(client: client).call + puts "Path MediaMTX rimossi: #{result.removed} (saltati: #{result.skipped})" puts "Fatto." end -end -def cleanup_orphan_mediamtx_paths(client, online_paths) - removed = 0 - - client.list_paths.each do |item| - name = item["name"] - next unless name&.start_with?("match_") - - session_id = name.delete_prefix("match_") - session = StreamSession.find_by(id: session_id) - should_remove = !online_paths.include?(name) && - (session.nil? || %w[ended error].include?(session.status)) - - next unless should_remove - - client.delete_path_name(name) - removed += 1 - puts " deleted path #{name}" - rescue Mediamtx::Client::Error, Faraday::Error => e - puts " skip #{name}: #{e.message}" + desc "Rimuove path MediaMTX orfani (dirette già terminate)" + task cleanup_orphan_paths: :environment do + result = Mediamtx::CleanupOrphanPaths.new.call + puts "Path MediaMTX rimossi: #{result.removed} (saltati: #{result.skipped})" end - - removed end diff --git a/backend/spec/services/mediamtx/cleanup_orphan_paths_spec.rb b/backend/spec/services/mediamtx/cleanup_orphan_paths_spec.rb new file mode 100644 index 0000000..a9b9ea2 --- /dev/null +++ b/backend/spec/services/mediamtx/cleanup_orphan_paths_spec.rb @@ -0,0 +1,105 @@ +require "rails_helper" + +RSpec.describe Mediamtx::CleanupOrphanPaths do + let!(:club) { Club.create!(name: "Club", sport: "volleyball", primary_color: "#e53935", secondary_color: "#ffffff") } + let!(:team) { club.teams.create!(name: "Team", sport: "volleyball") } + let!(:user) { User.create!(email: "coach@test.com", name: "Coach", password: "password123", role: "coach") } + let!(:match) { team.matches.create!(opponent_name: "Rival", scheduled_at: 1.day.from_now) } + + let(:client) { instance_double(Mediamtx::Client) } + + def path_item(name, online: false, ready: true) + { "name" => name, "online" => online, "ready" => ready } + end + + before do + allow(client).to receive(:online_path_names).and_return(Set.new) + allow(client).to receive(:delete_path) + allow(client).to receive(:delete_path_name) + end + + it "rimuove path live/match_{uuid} di sessioni terminate" do + session = StreamSession.create!( + match: match, user: user, platform: "matchlivetv", status: "ended", ended_at: 1.hour.ago + ) + allow(client).to receive(:list_paths).and_return([ + path_item(session.mediamtx_path_name) + ]) + + result = described_class.new(client: client).call + + expect(result.removed).to eq(1) + expect(client).to have_received(:delete_path).with(session) + end + + it "rimuove path orfani senza sessione nel database" do + orphan_id = SecureRandom.uuid + allow(client).to receive(:list_paths).and_return([ + path_item("live/match_#{orphan_id}") + ]) + + result = described_class.new(client: client).call + + expect(result.removed).to eq(1) + expect(client).to have_received(:delete_path_name).with("live/match_#{orphan_id}") + expect(client).to have_received(:delete_path_name).with("live/match_#{orphan_id}_air") + end + + it "non rimuove path con publisher online" do + session = StreamSession.create!( + match: match, user: user, platform: "matchlivetv", status: "ended", ended_at: 1.hour.ago + ) + allow(client).to receive(:online_path_names).and_return(Set.new([session.mediamtx_path_name])) + allow(client).to receive(:list_paths).and_return([ + path_item(session.mediamtx_path_name, online: true) + ]) + + result = described_class.new(client: client).call + + expect(result.removed).to eq(0) + expect(client).not_to have_received(:delete_path) + end + + it "non rimuove path di sessioni ancora live" do + session = StreamSession.create!( + match: match, user: user, platform: "matchlivetv", status: "live" + ) + allow(client).to receive(:list_paths).and_return([ + path_item(session.mediamtx_path_name) + ]) + + result = described_class.new(client: client).call + + expect(result.removed).to eq(0) + expect(client).not_to have_received(:delete_path) + end + + it "non rimuove path con replay in elaborazione recente" do + session = StreamSession.create!( + match: match, user: user, platform: "matchlivetv", status: "ended", ended_at: 30.minutes.ago + ) + Recording.create!( + stream_session: session, team: team, status: "processing", privacy_status: "public" + ) + allow(client).to receive(:list_paths).and_return([ + path_item(session.mediamtx_path_name) + ]) + + result = described_class.new(client: client).call + + expect(result.removed).to eq(0) + expect(client).not_to have_received(:delete_path) + end + + it "ignora path che non sono live/match_{uuid}" do + allow(client).to receive(:list_paths).and_return([ + path_item("all_others"), + path_item("match_legacy_without_live_prefix") + ]) + + result = described_class.new(client: client).call + + expect(result.removed).to eq(0) + expect(result.skipped).to eq(2) + end +end diff --git a/docs/ARCHITECTURE.md b/docs/ARCHITECTURE.md index 06f17a6..7633f48 100644 --- a/docs/ARCHITECTURE.md +++ b/docs/ARCHITECTURE.md @@ -413,3 +413,4 @@ cd infra && cp .env.example .env && docker compose up -d --build | 2026-06 | HLS live: edge → MediaMTX (fuori da Puma) | | 2026-06 | Ops: check `http_rails` + `http_public` separati, latenza p95 `UpLatencyTracker` | | 2026-06 | `RAILS_MAX_THREADS=5`; relay YouTube legge `mediamtx:8888` diretto | +| 2026-06 | Cleanup path MediaMTX orfani (`Mediamtx::CleanupOrphanPaths`); delete path sempre a fine diretta | diff --git a/docs/OPS_MONITORING.md b/docs/OPS_MONITORING.md index b0a60fa..dbd0809 100644 --- a/docs/OPS_MONITORING.md +++ b/docs/OPS_MONITORING.md @@ -101,6 +101,7 @@ cd /opt/matchlivetv/infra && bash scripts/install_production_cron.sh |----------|------| | ogni 3 min (Sidekiq) | `Ops::HealthMonitorJob` | | `*/10 * * * *` | `scan_ops_logs.sh` → `ops:ingest_logs` | +| `30 * * * *` | `recordings:cleanup_local` → segmenti disco + **path MediaMTX orfani** | ## Uptime Kuma (consigliato)