Elimina i path MediaMTX orfani dopo ogni diretta.
Corregge il cleanup dei path live/match_{uuid}, rimuove sempre il path a fine sessione e lo esegue ogni ora via cron.
Co-authored-by: Cursor <cursoragent@cursor.com>
This commit is contained in:
@@ -5,9 +5,11 @@ module Recordings
|
|||||||
|
|
||||||
def perform
|
def perform
|
||||||
result = Recordings::CleanupLocal.new.call
|
result = Recordings::CleanupLocal.new.call
|
||||||
|
orphan = Mediamtx::CleanupOrphanPaths.new.call
|
||||||
Rails.logger.info(
|
Rails.logger.info(
|
||||||
"[Recordings::CleanupLocalJob] removed=#{result.removed} " \
|
"[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
|
||||||
end
|
end
|
||||||
|
|||||||
94
backend/app/services/mediamtx/cleanup_orphan_paths.rb
Normal file
94
backend/app/services/mediamtx/cleanup_orphan_paths.rb
Normal file
@@ -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
|
||||||
@@ -12,11 +12,8 @@ module Sessions
|
|||||||
complete_youtube_broadcast! if @session.youtube_broadcast_id.present?
|
complete_youtube_broadcast! if @session.youtube_broadcast_id.present?
|
||||||
|
|
||||||
recording = Recordings::FinalizeSession.new(@session).call
|
recording = Recordings::FinalizeSession.new(@session).call
|
||||||
if recording
|
Recordings::UploadJob.perform_async(@session.id) if recording
|
||||||
Recordings::UploadJob.perform_async(@session.id)
|
remove_mediamtx_paths!
|
||||||
else
|
|
||||||
Mediamtx::Client.new.delete_path(@session)
|
|
||||||
end
|
|
||||||
|
|
||||||
log_event("ended")
|
log_event("ended")
|
||||||
SessionChannel.broadcast_message(@session, { type: "stream_event", 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: {})
|
@session.stream_events.create!(event_type: type, occurred_at: Time.current, metadata: {})
|
||||||
end
|
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!
|
def complete_youtube_broadcast!
|
||||||
Youtube::BroadcastService.new(@session.match.team).complete_broadcast!(@session.youtube_broadcast_id)
|
Youtube::BroadcastService.new(@session.match.team).complete_broadcast!(@session.youtube_broadcast_id)
|
||||||
rescue Youtube::BroadcastService::Error => e
|
rescue Youtube::BroadcastService::Error => e
|
||||||
|
|||||||
@@ -16,5 +16,8 @@ namespace :recordings do
|
|||||||
result = Recordings::CleanupLocal.new.call
|
result = Recordings::CleanupLocal.new.call
|
||||||
puts "Pulizia segmenti locali: rimossi=#{result.removed} " \
|
puts "Pulizia segmenti locali: rimossi=#{result.removed} " \
|
||||||
"liberati=#{result.freed_bytes} byte saltati=#{result.skipped}"
|
"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
|
||||||
end
|
end
|
||||||
|
|||||||
@@ -19,32 +19,14 @@ namespace :streams do
|
|||||||
puts " ended #{session.id} (fallback: #{e.message})"
|
puts " ended #{session.id} (fallback: #{e.message})"
|
||||||
end
|
end
|
||||||
|
|
||||||
removed = cleanup_orphan_mediamtx_paths(client, online)
|
result = Mediamtx::CleanupOrphanPaths.new(client: client).call
|
||||||
puts "Path MediaMTX rimossi: #{removed}"
|
puts "Path MediaMTX rimossi: #{result.removed} (saltati: #{result.skipped})"
|
||||||
puts "Fatto."
|
puts "Fatto."
|
||||||
end
|
end
|
||||||
end
|
|
||||||
|
|
||||||
def cleanup_orphan_mediamtx_paths(client, online_paths)
|
desc "Rimuove path MediaMTX orfani (dirette già terminate)"
|
||||||
removed = 0
|
task cleanup_orphan_paths: :environment do
|
||||||
|
result = Mediamtx::CleanupOrphanPaths.new.call
|
||||||
client.list_paths.each do |item|
|
puts "Path MediaMTX rimossi: #{result.removed} (saltati: #{result.skipped})"
|
||||||
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}"
|
|
||||||
end
|
end
|
||||||
|
|
||||||
removed
|
|
||||||
end
|
end
|
||||||
|
|||||||
105
backend/spec/services/mediamtx/cleanup_orphan_paths_spec.rb
Normal file
105
backend/spec/services/mediamtx/cleanup_orphan_paths_spec.rb
Normal file
@@ -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
|
||||||
@@ -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 | HLS live: edge → MediaMTX (fuori da Puma) |
|
||||||
| 2026-06 | Ops: check `http_rails` + `http_public` separati, latenza p95 `UpLatencyTracker` |
|
| 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 | `RAILS_MAX_THREADS=5`; relay YouTube legge `mediamtx:8888` diretto |
|
||||||
|
| 2026-06 | Cleanup path MediaMTX orfani (`Mediamtx::CleanupOrphanPaths`); delete path sempre a fine diretta |
|
||||||
|
|||||||
@@ -101,6 +101,7 @@ cd /opt/matchlivetv/infra && bash scripts/install_production_cron.sh
|
|||||||
|----------|------|
|
|----------|------|
|
||||||
| ogni 3 min (Sidekiq) | `Ops::HealthMonitorJob` |
|
| ogni 3 min (Sidekiq) | `Ops::HealthMonitorJob` |
|
||||||
| `*/10 * * * *` | `scan_ops_logs.sh` → `ops:ingest_logs` |
|
| `*/10 * * * *` | `scan_ops_logs.sh` → `ops:ingest_logs` |
|
||||||
|
| `30 * * * *` | `recordings:cleanup_local` → segmenti disco + **path MediaMTX orfani** |
|
||||||
|
|
||||||
## Uptime Kuma (consigliato)
|
## Uptime Kuma (consigliato)
|
||||||
|
|
||||||
|
|||||||
Reference in New Issue
Block a user