Timeout when linking CMAF muxer to sink while resuming pipeline

[zcesur]

Seeing intermittent timeouts during pad linking between CMAF muxer and sink. This eventually leads to a pattern match crash in the CMAF muxer. Hard to reproduce consistently but affects both audio/video tracks. Seems to be caused by the GenServer call in ‎Membrane.Core.Element.PadController.do_handle_link‎ which times out after 5 seconds

Could there be a deadlock or race condition during playback state transitions that's causing the sink to not respond to link requests? 🤔

[debug] <0.788.0>/:sink/{:cmaf_muxer, "audio_master"} Tried to link pad {Membrane.Pad, :output, #Reference<0.1819401566.4031774723.74705>}, but neighbour :sink
is not alive.

[debug] <0.788.0>/:sink/ Failed to establish link between {:cmaf_muxer, "audio_master"} via {Membrane.Pad, :output, #Reference<0.1819401566.4031774723.74705>} and
:sink via {Membrane.Pad, :input, "audio_master"} because {:cmaf_muxer, "audio_master"} is down.

[debug] <0.788.0>/:sink/ Spec #Reference<0.1819401566.4031774722.86411> status changed to linked internally
[debug] <0.788.0>/:sink/ Sending link response, link_id: #Reference<0.1819401566.4031774724.83906>, pad: {Membrane.Pad, :input, "audio_master"}
[debug] <0.788.0>/:sink/{:cmaf_muxer, "audio_master"} Ignoring unlinking pad {Membrane.Pad, :output, #Reference<0.1819401566.4031774723.74705>} that hasn't been successfully linked
[debug] <0.788.0>/:sink/ Spec #Reference<0.1819401566.4031774722.86411> status changed to linking externally
[debug] <0.788.0>/:sink/ Proceeding spec #Reference<0.1819401566.4031774723.74707> startup: initializing, dependent specs: []
[debug] <0.788.0>/:sink/ Proceeding spec #Reference<0.1819401566.4031774723.74707> startup: initializing, dependent specs: []
[debug] <0.788.0>/:sink/ Spec #Reference<0.1819401566.4031774723.74707> status changed to initialized
[debug] <0.788.0>/:sink/ Got internal link request, pad ref {Membrane.Pad, :input, "video_master"}, child {:parser, "video_master"}, spec #Reference<0.1819401566.4031774723.74707>
[debug] <0.788.0>/:sink/ Spec #Reference<0.1819401566.4031774723.74707> status changed to linking internally
[debug] <0.788.0>/:sink/{:parser, "video_master"} Element handle link on pad :output with pad {Membrane.Pad, :input, #Reference<0.1819401566.4031774723.74731>} of child {:cmaf_muxer, "video_master"}
[debug] <0.788.0>/:sink/{:cmaf_muxer, "video_master"} Element handle link on pad {Membrane.Pad, :input, #Reference<0.1819401566.4031774723.74731>} with pad :output of child {:parser, "video_master"}
[debug] <0.788.0>/:sink/{:cmaf_muxer, "video_master"} Element handle link on pad {Membrane.Pad, :output, #Reference<0.1819401566.4031774723.74732>} with pad {Membrane.Pad, :input, "video_master"} of child :sink
[debug] <0.788.0>/:sink/ Failed to establish link between {:cmaf_muxer, "video_master"} via {Membrane.Pad, :output, #Reference<0.1819401566.4031774723.74732>} and
:sink via {Membrane.Pad, :input, "video_master"} because {:cmaf_muxer, "video_master"} is down.

[debug] <0.788.0>/:sink/ Spec #Reference<0.1819401566.4031774723.74707> status changed to linked internally
[debug] <0.788.0>/:sink/ Sending link response, link_id: #Reference<0.1819401566.4031774724.83907>, pad: {Membrane.Pad, :input, "video_master"}
[debug] <0.788.0>/:sink/ Spec #Reference<0.1819401566.4031774723.74707> status changed to linking externally
[debug] <0.788.0>/:tee_audio Element handle link on pad {Membrane.Pad, :output, #Reference<0.1819401566.4031774724.83908>} with pad {Membrane.Pad, :input, "audio_master"} of child :sink
[debug] <0.788.0>/:sink/{:cmaf_muxer, "video_master"} Tried to link pad {Membrane.Pad, :output, #Reference<0.1819401566.4031774723.74732>}, but neighbour :sink
is not alive.

[debug] <0.788.0>/:sink/{:cmaf_muxer, "video_master"} Ignoring unlinking pad {Membrane.Pad, :output, #Reference<0.1819401566.4031774723.74732>} that hasn't been successfully linked
[debug] <0.788.0>/:sink/ Handle link %Membrane.Core.Parent.Link.Endpoint{
  child: :sink,
  pad_spec: {Membrane.Pad, :input, "audio_master"},
  pad_ref: {Membrane.Pad, :input, "audio_master"},
  pid: #PID<0.824.0>,
  pad_props: %{
    options: [
      track_name: "audio_master",
      encoding: :AAC,
      segment_duration: 6000000000,
      partial_segment_duration: 1000000000
    ],
    toilet_capacity: nil,
    target_queue_size: nil,
    min_demand_factor: nil,
    auto_demand_size: nil,
    throttling_factor: nil
  },
  pad_info: %{
    name: :input,
    options: [
      encoding: [
        spec: ":AAC | :H264 | :H265",
        description: "Encoding type determining which parser will be used for the given stream.\n"
      ],
      track_name: [
        spec: "String.t() | nil",
        default: nil,
        description: "Name that will be used to name the media playlist for the given track, as well as its header and segments files.\nIt must not contain any URI reserved characters\n"
      ],
      segment_duration: [
        spec: "Membrane.Time.t()",
        description: "The minimal segment duration of the regular segments.\n"
      ],
      partial_segment_duration: [
        spec: "Membrane.Time.t() | nil",
        default: nil,
        description: "The segment duration of the partial segments.\nIf not set then the bin won't produce any partial segments.\n"
      ],
      max_framerate: [
        spec: "float() | nil",
        default: nil,
        description: "The maximal framerate of video variant. This information is used in master playlist.\n\nWhen set to nil then this information won't be added to master playlist. For audio it should be set to nil.\n"
      ]
    ],
    direction: :input,
    accepted_formats_str: ["Membrane.AAC", "Membrane.H264", "Membrane.H265"],
    availability: :on_request
  },
  child_spec_ref: #Reference<0.1819401566.4031774725.64144>
}
[debug] <0.788.0>/:sink/{:parser, "audio_master"} Element handle link on pad :input with pad {Membrane.Pad, :output, #Reference<0.1819401566.4031774724.83908>} of child :tee_audio
[debug] <0.788.0>/:sink/ Spec #Reference<0.1819401566.4031774722.86411> status changed to ready
[debug] <0.788.0>/:sink/{:parser, "audio_master"} Got play request
[debug] <0.788.0>/:sink/ Cleaning spec %{status: :linking_externally, children_names: MapSet.new([cmaf_muxer: "audio_master", parser: "audio_master"]), awaiting_responses: MapSet.new([]), links_ids: [#Reference<0.1819401566.4031774722.86412>, #Reference<0.1819401566.4031774722.86413>, #Reference<0.1819401566.4031774722.86414>], dependent_specs: MapSet.new([])}
[debug] <0.788.0>/:sink/{:cmaf_muxer, "audio_master"} Got play request
[debug] <0.788.0>/:tee_audio Sending stream format through pad {Membrane.Pad, :output, #Reference<0.1819401566.4031774724.83908>}
Stream format: %Membrane.AAC{sample_rate: 48000, channels: 2, profile: :LC, mpeg_version: 4, samples_per_frame: 1024, frames_per_buffer: 1, encapsulation: :none, config: nil}

[debug] <0.788.0>/:tee_video Element handle link on pad {Membrane.Pad, :output, #Reference<0.1819401566.4031774724.83909>} with pad {Membrane.Pad, :input, "video_master"} of child :sink
[debug] <0.788.0>/:sink/{:parser, "audio_master"} Sending stream format through pad :output
Stream format: %Membrane.AAC{sample_rate: 48000, channels: 2, profile: :LC, mpeg_version: 4, samples_per_frame: 1024, frames_per_buffer: 1, encapsulation: :none, config: {:esds, <<3, 128, 128, 128, 34, 0, 1, 0, 4, 128, 128, 128, 20, 64, 21, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 5, 128, 128, 128, 2, 17, 144, 6, 128, 128, 128, 1, 2>>}}

[debug] <0.788.0>/:sink/ Handle link %Membrane.Core.Parent.Link.Endpoint{
  child: :sink,
  pad_spec: {Membrane.Pad, :input, "video_master"},
  pad_ref: {Membrane.Pad, :input, "video_master"},
  pid: #PID<0.824.0>,
  pad_props: %{
    options: [
      track_name: "video_master",
      encoding: :H264,
      segment_duration: 6000000000,
      partial_segment_duration: 1000000000
    ],
    toilet_capacity: nil,
    target_queue_size: nil,
    min_demand_factor: nil,
    auto_demand_size: nil,
    throttling_factor: nil
  },
  pad_info: %{
    name: :input,
    options: [
      encoding: [
        spec: ":AAC | :H264 | :H265",
        description: "Encoding type determining which parser will be used for the given stream.\n"
      ],
      track_name: [
        spec: "String.t() | nil",
        default: nil,
        description: "Name that will be used to name the media playlist for the given track, as well as its header and segments files.\nIt must not contain any URI reserved characters\n"
      ],
      segment_duration: [
        spec: "Membrane.Time.t()",
        description: "The minimal segment duration of the regular segments.\n"
      ],
      partial_segment_duration: [
        spec: "Membrane.Time.t() | nil",
        default: nil,
        description: "The segment duration of the partial segments.\nIf not set then the bin won't produce any partial segments.\n"
      ],
      max_framerate: [
        spec: "float() | nil",
        default: nil,
        description: "The maximal framerate of video variant. This information is used in master playlist.\n\nWhen set to nil then this information won't be added to master playlist. For audio it should be set to nil.\n"
      ]
    ],
    direction: :input,
    accepted_formats_str: ["Membrane.AAC", "Membrane.H264", "Membrane.H265"],
    availability: :on_request
  },
  child_spec_ref: #Reference<0.1819401566.4031774725.64144>
}
[debug] <0.788.0>/:sink/{:parser, "video_master"} Element handle link on pad :input with pad {Membrane.Pad, :output, #Reference<0.1819401566.4031774724.83909>} of child :tee_video
[debug] <0.788.0>/:sink/ Spec #Reference<0.1819401566.4031774723.74707> status changed to ready
[debug] <0.788.0>/:sink/{:parser, "video_master"} Got play request
[debug] <0.788.0>/:sink/{:cmaf_muxer, "video_master"} Got play request
[debug] <0.788.0>/:sink/{:parser, "video_master"} Transiting `flow_control: :auto` pads to :pull effective flow control
[debug] <0.788.0>/:sink/ Cleaning spec %{status: :linking_externally, children_names: MapSet.new([cmaf_muxer: "video_master", parser: "video_master"]), awaiting_responses: MapSet.new([]), links_ids: [#Reference<0.1819401566.4031774721.97285>, #Reference<0.1819401566.4031774721.97286>, #Reference<0.1819401566.4031774721.97287>], dependent_specs: MapSet.new([])}
[debug] <0.788.0>/:tee_video Sending stream format through pad {Membrane.Pad, :output, #Reference<0.1819401566.4031774724.83909>}
Stream format: %Membrane.H264{width: 1920, height: 1080, profile: :high, alignment: :au, nalu_in_metadata?: true, framerate: nil, stream_structure: {:avc3, <<1, 100, 0, 40, 255, 224, 0>>}}

[debug] <0.788.0>/ Spec #Reference<0.1819401566.4031774724.83905> status changed to linked internally
[debug] <0.788.0>/ Spec #Reference<0.1819401566.4031774724.83905> status changed to ready
[debug] <0.788.0>/ Cleaning spec %{status: :linking_internally, children_names: MapSet.new([]), awaiting_responses: MapSet.new([]), links_ids: [#Reference<0.1819401566.4031774724.83906>, #Reference<0.1819401566.4031774724.83907>], dependent_specs: MapSet.new([])}
[error] GenServer #PID<0.2353.0> terminating
** (MatchError) no match of right hand side value: %{input: [{Membrane.Pad, :input, #Reference<0.1819401566.4031774723.74731>}]}
    (membrane_mp4_plugin 0.35.2) lib/membrane_mp4/muxer/cmaf.ex:227: Membrane.MP4.Muxer.CMAF.handle_playing/2
    (membrane_core 1.0.1) lib/membrane/core/callback_handler.ex:139: Membrane.Core.CallbackHandler.exec_callback/4
    (membrane_core 1.0.1) lib/membrane/core/callback_handler.ex:69: Membrane.Core.CallbackHandler.exec_and_handle_callback/5
    (membrane_core 1.0.1) lib/membrane/core/element/lifecycle_controller.ex:84: Membrane.Core.Element.LifecycleController.handle_playing/1
    (membrane_core 1.0.1) lib/membrane/core/element.ex:249: Membrane.Core.Element.handle_info/2
    (stdlib 6.1) gen_server.erl:2345: :gen_server.try_handle_info/3
    (stdlib 6.1) gen_server.erl:2433: :gen_server.handle_msg/6
    (stdlib 6.1) proc_lib.erl:329: :proc_lib.init_p_do_apply/3
Last message: {Membrane.Core.Message, :play, [], []}
State: %Membrane.Core.Element.State{module: Membrane.MP4.Muxer.CMAF, name: {:cmaf_muxer, "video_master"}, parent_pid: #PID<0.824.0>, playback: :stopped, type: :filter, internal_state: %{segment_min_duration: 6000000000, video_pad: nil, pad_to_track_data: %{}, all_input_pads_ready?: false, buffers_awaiting_init: [], sample_queues: %{}, finish_current_segment?: false, awaiting_stream_formats: %{}, pads_registration_order: [{Membrane.Pad, :input, #Reference<0.1819401566.4031774723.74731>}], chunk_duration_range: %Membrane.MP4.Muxer.CMAF.DurationRange{min: 50000000, target: 1000000000}}, pad_refs: [{Membrane.Pad, :input, #Reference<0.1819401566.4031774723.74731>}], pads_info: %{input: %{name: :input, options: [], direction: :input, accepted_formats_str: ["%AAC{config: {:esds, _esds}}", "%Opus{self_delimiting?: false}", "%H264{stream_structure: structure, alignment: :au} when H264.is_avc(structure)", "%H265{stream_structure: structure, alignment: :au} when H265.is_hvc(structure)"], availability: :on_request, flow_control: :manual, demand_unit: :buffers}, output: %{name: :output, options: [tracks: [spec: "[Membrane.Pad.dynamic_id()] | :all", default: :all, description: "A list of the input pad ids that should be muxed together into a single output track.\n\nIf not specified the pad will include all unreferenced input pads.\n"]], direction: :output, accepted_formats_str: ["Membrane.CMAF.Track"], availability: :on_request, flow_control: :manual, demand_unit: nil}}, synchronization: %{timers: %{}, clock: nil, stream_sync: :membrane_no_sync, parent_clock: #PID<0.825.0>, latency: 0}, delayed_demands: MapSet.new([]), effective_flow_control: :push, initialized?: true, terminating?: false, setup_incomplete?: false, supplying_demand?: false, handling_action?: false, stalker: %Membrane.Core.Stalker{pid: #PID<0.789.0>, ets: #Reference<0.1819401566.4031905793.76364>}, resource_guard: #PID<0.2354.0>, subprocess_supervisor: #PID<0.2352.0>, handle_demand_loop_counter: 0, demand_size: nil, pads_to_snapshot: MapSet.new([]), playback_queue: [], pads_data: %{{Membrane.Pad, :input, #Reference<0.1819401566.4031774723.74731>} => %Membrane.Element.PadData{availability: :on_request, stream_format: nil, direction: :input, flow_control: :manual, name: :input, ref: {Membrane.Pad, :input, #Reference<0.1819401566.4031774723.74731>}, options: %{}, pid: #PID<0.2347.0>, other_ref: :output, input_queue: %Membrane.Core.Element.InputQueue{q: #Qex<[]>, log_tag: "{Membrane.Pad, :input, #Reference<0.1819401566.4031774723.74731>}", target_size: 40, atomic_demand: %Membrane.Core.Element.AtomicDemand{counter: %Membrane.Core.Element.AtomicDemand.DistributedAtomic{worker: #PID<0.2435.0>, atomic_ref: #Reference<0.1819401566.4031905793.98144>}, receiver_status: %Membrane.Core.Element.AtomicDemand.DistributedAtomic{worker: #PID<0.2436.0>, atomic_ref: #Reference<0.1819401566.4031905793.98146>}, receiver_process: #PID<0.2353.0>, sender_status: %Membrane.Core.Element.AtomicDemand.DistributedAtomic{worker: #PID<0.2437.0>, atomic_ref: #Reference<0.1819401566.4031905793.98148>}, sender_process: #PID<0.2347.0>, sender_pad_ref: :output, throttling_factor: 1, toilet_capacity: 200, receiver_demand_unit: :buffers, buffered_decrementation: 0, toilet_overflowed?: false}, inbound_metric: Membrane.Buffer.Metric.Count, outbound_metric: Membrane.Buffer.Metric.Count, pad_ref: {Membrane.Pad, :input, #Reference<0.1819401566.4031774723.74731>}, stalker_metrics: %{size: #Reference<0.1819401566.4031905793.98151>}, size: 0, demand: 40}, demand: 0, incoming_demand: nil, demand_unit: :buffers, start_of_stream?: false, end_of_stream?: false, auto_demand_size: nil, sticky_messages: [], atomic_demand: %Membrane.Core.Element.AtomicDemand{counter: %Membrane.Core.Element.AtomicDemand.DistributedAtomic{worker: #PID<0.2435.0>, atomic_ref: #Reference<0.1819401566.4031905793.98144>}, receiver_status: %Membrane.Core.Element.AtomicDemand.DistributedAtomic{worker: #PID<0.2436.0>, atomic_ref: #Reference<0.1819401566.4031905793.98146>}, receiver_process: #PID<0.2353.0>, sender_status: %Membrane.Core.Element.AtomicDemand.DistributedAtomic{worker: #PID<0.2437.0>, atomic_ref: #Reference<0.1819401566.4031905793.98148>}, sender_process: #PID<0.2347.0>, sender_pad_ref: :output, throttling_factor: 1, ...}, manual_demand_size: 0, associated_pads: [], sticky_events: [], stream_format_validation_params: [], other_demand_unit: :buffers, other_effective_flow_control: :push, stalker_metrics: %{...}, ...}}}
[error] GenServer #PID<0.2341.0> terminating
** (MatchError) no match of right hand side value: %{input: [{Membrane.Pad, :input, #Reference<0.1819401566.4031774723.74704>}]}
    (membrane_mp4_plugin 0.35.2) lib/membrane_mp4/muxer/cmaf.ex:227: Membrane.MP4.Muxer.CMAF.handle_playing/2
    (membrane_core 1.0.1) lib/membrane/core/callback_handler.ex:139: Membrane.Core.CallbackHandler.exec_callback/4
    (membrane_core 1.0.1) lib/membrane/core/callback_handler.ex:69: Membrane.Core.CallbackHandler.exec_and_handle_callback/5
    (membrane_core 1.0.1) lib/membrane/core/element/lifecycle_controller.ex:84: Membrane.Core.Element.LifecycleController.handle_playing/1
    (membrane_core 1.0.1) lib/membrane/core/element.ex:249: Membrane.Core.Element.handle_info/2
    (stdlib 6.1) gen_server.erl:2345: :gen_server.try_handle_info/3
    (stdlib 6.1) gen_server.erl:2433: :gen_server.handle_msg/6
    (stdlib 6.1) proc_lib.erl:329: :proc_lib.init_p_do_apply/3
Last message: {Membrane.Core.Message, :play, [], []}
State: %Membrane.Core.Element.State{module: Membrane.MP4.Muxer.CMAF, name: {:cmaf_muxer, "audio_master"}, parent_pid: #PID<0.824.0>, playback: :stopped, type: :filter, internal_state: %{segment_min_duration: 6000000000, video_pad: nil, pad_to_track_data: %{}, all_input_pads_ready?: false, buffers_awaiting_init: [], sample_queues: %{}, finish_current_segment?: false, awaiting_stream_formats: %{}, pads_registration_order: [{Membrane.Pad, :input, #Reference<0.1819401566.4031774723.74704>}], chunk_duration_range: %Membrane.MP4.Muxer.CMAF.DurationRange{min: 50000000, target: 1000000000}}, pad_refs: [{Membrane.Pad, :input, #Reference<0.1819401566.4031774723.74704>}], pads_info: %{input: %{name: :input, options: [], direction: :input, accepted_formats_str: ["%AAC{config: {:esds, _esds}}", "%Opus{self_delimiting?: false}", "%H264{stream_structure: structure, alignment: :au} when H264.is_avc(structure)", "%H265{stream_structure: structure, alignment: :au} when H265.is_hvc(structure)"], availability: :on_request, flow_control: :manual, demand_unit: :buffers}, output: %{name: :output, options: [tracks: [spec: "[Membrane.Pad.dynamic_id()] | :all", default: :all, description: "A list of the input pad ids that should be muxed together into a single output track.\n\nIf not specified the pad will include all unreferenced input pads.\n"]], direction: :output, accepted_formats_str: ["Membrane.CMAF.Track"], availability: :on_request, flow_control: :manual, demand_unit: nil}}, synchronization: %{timers: %{}, clock: nil, stream_sync: :membrane_no_sync, parent_clock: #PID<0.825.0>, latency: 0}, delayed_demands: MapSet.new([]), effective_flow_control: :push, initialized?: true, terminating?: false, setup_incomplete?: false, supplying_demand?: false, handling_action?: false, stalker: %Membrane.Core.Stalker{pid: #PID<0.789.0>, ets: #Reference<0.1819401566.4031905793.76364>}, resource_guard: #PID<0.2342.0>, subprocess_supervisor: #PID<0.2340.0>, handle_demand_loop_counter: 0, demand_size: nil, pads_to_snapshot: MapSet.new([]), playback_queue: [], pads_data: %{{Membrane.Pad, :input, #Reference<0.1819401566.4031774723.74704>} => %Membrane.Element.PadData{availability: :on_request, stream_format: nil, direction: :input, flow_control: :manual, name: :input, ref: {Membrane.Pad, :input, #Reference<0.1819401566.4031774723.74704>}, options: %{}, pid: #PID<0.2335.0>, other_ref: :output, input_queue: %Membrane.Core.Element.InputQueue{q: #Qex<[]>, log_tag: "{Membrane.Pad, :input, #Reference<0.1819401566.4031774723.74704>}", target_size: 40, atomic_demand: %Membrane.Core.Element.AtomicDemand{counter: %Membrane.Core.Element.AtomicDemand.DistributedAtomic{worker: #PID<0.2361.0>, atomic_ref: #Reference<0.1819401566.4031905793.97380>}, receiver_status: %Membrane.Core.Element.AtomicDemand.DistributedAtomic{worker: #PID<0.2362.0>, atomic_ref: #Reference<0.1819401566.4031905793.97383>}, receiver_process: #PID<0.2341.0>, sender_status: %Membrane.Core.Element.AtomicDemand.DistributedAtomic{worker: #PID<0.2363.0>, atomic_ref: #Reference<0.1819401566.4031905793.97386>}, sender_process: #PID<0.2335.0>, sender_pad_ref: :output, throttling_factor: 1, toilet_capacity: 200, receiver_demand_unit: :buffers, buffered_decrementation: 0, toilet_overflowed?: false}, inbound_metric: Membrane.Buffer.Metric.Count, outbound_metric: Membrane.Buffer.Metric.Count, pad_ref: {Membrane.Pad, :input, #Reference<0.1819401566.4031774723.74704>}, stalker_metrics: %{size: #Reference<0.1819401566.4031905793.97390>}, size: 0, demand: 40}, demand: 0, incoming_demand: nil, demand_unit: :buffers, start_of_stream?: false, end_of_stream?: false, auto_demand_size: nil, sticky_messages: [], atomic_demand: %Membrane.Core.Element.AtomicDemand{counter: %Membrane.Core.Element.AtomicDemand.DistributedAtomic{worker: #PID<0.2361.0>, atomic_ref: #Reference<0.1819401566.4031905793.97380>}, receiver_status: %Membrane.Core.Element.AtomicDemand.DistributedAtomic{worker: #PID<0.2362.0>, atomic_ref: #Reference<0.1819401566.4031905793.97383>}, receiver_process: #PID<0.2341.0>, sender_status: %Membrane.Core.Element.AtomicDemand.DistributedAtomic{worker: #PID<0.2363.0>, atomic_ref: #Reference<0.1819401566.4031905793.97386>}, sender_process: #PID<0.2335.0>, sender_pad_ref: :output, throttling_factor: 1, ...}, manual_demand_size: 0, associated_pads: [], sticky_events: [], stream_format_validation_params: [], other_demand_unit: :buffers, other_effective_flow_control: :pull, stalker_metrics: %{...}, ...}}}
[debug] <0.788.0>/:sink/{:cmaf_muxer, "audio_master"} subprocess supervisor got exit request from parent, reason: :shutdown, shutting down children
[debug] <0.788.0>/:sink/ Child {:cmaf_muxer, "audio_master"} crashed but was not a member of any crash group.
Terminating.

[debug] <0.788.0>/:sink/{:cmaf_muxer, "video_master"} subprocess supervisor got exit request from parent, reason: :shutdown, shutting down children
[error] <0.788.0>/:sink/ Terminating with reason: {:membrane_child_crash, {:cmaf_muxer, "audio_master"}, {{:badmatch, %{input: [{Membrane.Pad, :input, #Reference<0.1819401566.4031774723.74704>}]}}, [{Membrane.MP4.Muxer.CMAF, :handle_playing, 2, [file: ~c"lib/membrane_mp4/muxer/cmaf.ex", line: 227]}, {Membrane.Core.CallbackHandler, :exec_callback, 4, [file: ~c"lib/membrane/core/callback_handler.ex", line: 139]}, {Membrane.Core.CallbackHandler, :exec_and_handle_callback, 5, [file: ~c"lib/membrane/core/callback_handler.ex", line: 69]}, {Membrane.Core.Element.LifecycleController, :handle_playing, 1, [file: ~c"lib/membrane/core/element/lifecycle_controller.ex", line: 84]}, {Membrane.Core.Element, :handle_info, 2, [file: ~c"lib/membrane/core/element.ex", line: 249]}, {:gen_server, :try_handle_info, 3, [file: ~c"gen_server.erl", line: 2345]}, {:gen_server, :handle_msg, 6, [file: ~c"gen_server.erl", line: 2433]}, {:proc_lib, :init_p_do_apply, 3, [file: ~c"proc_lib.erl", line: 329]}]}}
[debug] <0.788.0>/:sink/ subprocess supervisor got exit request from parent, reason: :shutdown, shutting down children
[debug] <0.788.0>/:sink/{:parser, "video_master"} subprocess supervisor got exit request from parent, reason: :shutdown, shutting down children
[debug] <0.788.0>/:sink/:sink subprocess supervisor got exit request from parent, reason: :shutdown, shutting down children
[info] Elixir.Algora.Pipeline.Storage.Manifest terminating because of :shutdown
[debug] <0.788.0>/ Child :sink crashed but was not a member of any crash group.
Terminating.

[debug] <0.788.0>/:sink/{:parser, "audio_master"} subprocess supervisor got exit request from parent, reason: :shutdown, shutting down children
[error] <0.788.0>/ Terminating with reason: {:membrane_child_crash, :sink, {:membrane_child_crash, {:cmaf_muxer, "audio_master"}, {{:badmatch, %{input: [{Membrane.Pad, :input, #Reference<0.1819401566.4031774723.74704>}]}}, [{Membrane.MP4.Muxer.CMAF, :handle_playing, 2, [file: ~c"lib/membrane_mp4/muxer/cmaf.ex", line: 227]}, {Membrane.Core.CallbackHandler, :exec_callback, 4, [file: ~c"lib/membrane/core/callback_handler.ex", line: 139]}, {Membrane.Core.CallbackHandler, :exec_and_handle_callback, 5, [file: ~c"lib/membrane/core/callback_handler.ex", line: 69]}, {Membrane.Core.Element.LifecycleController, :handle_playing, 1, [file: ~c"lib/membrane/core/element/lifecycle_controller.ex", line: 84]}, {Membrane.Core.Element, :handle_info, 2, [file: ~c"lib/membrane/core/element.ex", line: 249]}, {:gen_server, :try_handle_info, 3, [file: ~c"gen_server.erl", line: 2345]}, {:gen_server, :handle_msg, 6, [file: ~c"gen_server.erl", line: 2433]}, {:proc_lib, :init_p_do_apply, 3, [file: ~c"proc_lib.erl", line: 329]}]}}}
[debug] <0.788.0>/ Pipeline supervisor got exit from pipeline with reason {:membrane_child_crash, :sink, {:membrane_child_crash, {:cmaf_muxer, "audio_master"}, {{:badmatch, %{input: [{Membrane.Pad, :input, #Reference<0.1819401566.4031774723.74704>}]}}, [{Membrane.MP4.Muxer.CMAF, :handle_playing, 2, [file: ~c"lib/membrane_mp4/muxer/cmaf.ex", line: 227]}, {Membrane.Core.CallbackHandler, :exec_callback, 4, [file: ~c"lib/membrane/core/callback_handler.ex", line: 139]}, {Membrane.Core.CallbackHandler, :exec_and_handle_callback, 5, [file: ~c"lib/membrane/core/callback_handler.ex", line: 69]}, {Membrane.Core.Element.LifecycleController, :handle_playing, 1, [file: ~c"lib/membrane/core/element/lifecycle_controller.ex", line: 84]}, {Membrane.Core.Element, :handle_info, 2, [file: ~c"lib/membrane/core/element.ex", line: 249]}, {:gen_server, :try_handle_info, 3, [file: ~c"gen_server.erl", line: 2345]}, {:gen_server, :handle_msg, 6, [file: ~c"gen_server.erl", line: 2433]}, {:proc_lib, :init_p_do_apply, 3, [file: ~c"proc_lib.erl", line: 329]}]}}}, stopping subprocess supervisor
[debug] <0.788.0>/ subprocess supervisor got exit request from parent, reason: :shutdown, shutting down children
[debug] <0.788.0>/:funnel_video subprocess supervisor got exit request from parent, reason: :shutdown, shutting down children
[debug] <0.788.0>/{:audio_reconnect, 2} subprocess supervisor got exit request from parent, reason: :shutdown, shutting down children
[debug] <0.788.0>/{:src, 2}/ subprocess supervisor got exit request from parent, reason: :shutdown, shutting down children
[debug] <0.788.0>/{:video_reconnect, 2} subprocess supervisor got exit request from parent, reason: :shutdown, shutting down children
[debug] <0.788.0>/{:src, 0}/ subprocess supervisor got exit request from parent, reason: :shutdown, shutting down children
[debug] <0.788.0>/{:src, 2}/:funnel_audio subprocess supervisor got exit request from parent, reason: :shutdown, shutting down children
[debug] <0.788.0>/{:src, 2}/:funnel_video subprocess supervisor got exit request from parent, reason: :shutdown, shutting down children
[debug] <0.788.0>/{:video_reconnect, 0} subprocess supervisor got exit request from parent, reason: :shutdown, shutting down children
[debug] <0.788.0>/{:src, 0}/:src subprocess supervisor got exit request from parent, reason: :shutdown, shutting down children
[debug] <0.788.0>/{:src, 0}/:audio_parser subprocess supervisor got exit request from parent, reason: :shutdown, shutting down children
[debug] <0.788.0>/{:src, 2}/:audio_parser subprocess supervisor got exit request from parent, reason: :shutdown, shutting down children
[debug] <0.788.0>/:funnel_audio subprocess supervisor got exit request from parent, reason: :shutdown, shutting down children
[debug] <0.788.0>/{:audio_reconnect, 0} subprocess supervisor got exit request from parent, reason: :shutdown, shutting down children
[debug] <0.788.0>/{:src, 1}/ subprocess supervisor got exit request from parent, reason: :shutdown, shutting down children
[debug] <0.788.0>/:tee_audio subprocess supervisor got exit request from parent, reason: :shutdown, shutting down children
[debug] <0.788.0>/{:audio_reconnect, 1} subprocess supervisor got exit request from parent, reason: :shutdown, shutting down children
[debug] <0.788.0>/{:src, 2}/:src subprocess supervisor got exit request from parent, reason: :shutdown, shutting down children
[debug] <0.788.0>/{:src, 1}/:src subprocess supervisor got exit request from parent, reason: :shutdown, shutting down children
[debug] <0.788.0>/{:src, 0}/:video_parser subprocess supervisor got exit request from parent, reason: :shutdown, shutting down children
[debug] <0.788.0>/{:src, 0}/:funnel_audio subprocess supervisor got exit request from parent, reason: :shutdown, shutting down children
[debug] <0.788.0>/{:src, 2}/:video_parser subprocess supervisor got exit request from parent, reason: :shutdown, shutting down children
[debug] <0.788.0>/:tee_video subprocess supervisor got exit request from parent, reason: :shutdown, shutting down children
[debug] <0.788.0>/{:video_reconnect, 1} subprocess supervisor got exit request from parent, reason: :shutdown, shutting down children
[debug] <0.788.0>/{:src, 0}/:funnel_video subprocess supervisor got exit request from parent, reason: :shutdown, shutting down children
[debug] <0.788.0>/{:src, 1}/:demuxer subprocess supervisor got exit request from parent, reason: :shutdown, shutting down children
[debug] <0.788.0>/{:src, 0}/:demuxer subprocess supervisor got exit request from parent, reason: :shutdown, shutting down children
[debug] <0.788.0>/{:src, 2}/:demuxer subprocess supervisor got exit request from parent, reason: :shutdown, shutting down children
[debug] <0.788.0>/{:src, 1}/:funnel_video subprocess supervisor got exit request from parent, reason: :shutdown, shutting down children
[debug] <0.788.0>/{:src, 1}/:audio_parser subprocess supervisor got exit request from parent, reason: :shutdown, shutting down children
[debug] <0.788.0>/{:src, 1}/:funnel_audio subprocess supervisor got exit request from parent, reason: :shutdown, shutting down children
[debug] <0.788.0>/{:src, 1}/:video_parser subprocess supervisor got exit request from parent, reason: :shutdown, shutting down children
[debug] <0.788.0>/ Pipeline supervisor got exit from subprocess supervisor, exiting
[error] GenServer #PID<0.785.0> terminating
** (stop) {:membrane_child_crash, :sink, {:membrane_child_crash, {:cmaf_muxer, "audio_master"}, {{:badmatch, %{input: [{Membrane.Pad, :input, #Reference<0.1819401566.4031774723.74704>}]}}, [{Membrane.MP4.Muxer.CMAF, :handle_playing, 2, [file: ~c"lib/membrane_mp4/muxer/cmaf.ex", line: 227]}, {Membrane.Core.CallbackHandler, :exec_callback, 4, [file: ~c"lib/membrane/core/callback_handler.ex", line: 139]}, {Membrane.Core.CallbackHandler, :exec_and_handle_callback, 5, [file: ~c"lib/membrane/core/callback_handler.ex", line: 69]}, {Membrane.Core.Element.LifecycleController, :handle_playing, 1, [file: ~c"lib/membrane/core/element/lifecycle_controller.ex", line: 84]}, {Membrane.Core.Element, :handle_info, 2, [file: ~c"lib/membrane/core/element.ex", line: 249]}, {:gen_server, :try_handle_info, 3, [file: ~c"gen_server.erl", line: 2345]}, {:gen_server, :handle_msg, 6, [file: ~c"gen_server.erl", line: 2433]}, {:proc_lib, :init_p_do_apply, 3, [file: ~c"proc_lib.erl", line: 329]}]}}}
Last message: {:EXIT, #PID<0.786.0>, {:membrane_child_crash, :sink, {:membrane_child_crash, {:cmaf_muxer, "audio_master"}, {{:badmatch, %{input: [{Membrane.Pad, :input, #Reference<0.1819401566.4031774723.74704>}]}}, [{Membrane.MP4.Muxer.CMAF, :handle_playing, 2, [file: ~c"lib/membrane_mp4/muxer/cmaf.ex", line: 227]}, {Membrane.Core.CallbackHandler, :exec_callback, 4, [file: ~c"lib/membrane/core/callback_handler.ex", line: 139]}, {Membrane.Core.CallbackHandler, :exec_and_handle_callback, 5, [file: ~c"lib/membrane/core/callback_handler.ex", line: 69]}, {Membrane.Core.Element.LifecycleController, :handle_playing, 1, [file: ~c"lib/membrane/core/element/lifecycle_controller.ex", line: 84]}, {Membrane.Core.Element, :handle_info, 2, [file: ~c"lib/membrane/core/element.ex", line: 249]}, {:gen_server, :try_handle_info, 3, [file: ~c"gen_server.erl", line: 2345]}, {:gen_server, :handle_msg, 6, [file: ~c"gen_server.erl", line: 2433]}, {:proc_lib, :init_p_do_apply, 3, [file: ~c"proc_lib.erl", line: 329]}]}}}}

[kaf-lamed-beyt]

How can i reproduce this @zcesur?

I'm running the project locally now and nothing seems to be broken.

[zcesur]

Yeah it's hard to reproduce consistently. Try start streaming -> stop streaming -> start streaming back to back a handful of times. Eventually it should crash

[kaf-lamed-beyt]

Oh okay. I'll let you know how it goes

[zcesur]

Are you using a private bucket? Looks like a permission error

Just a heads up - this particular issue is pretty deep in the video processing pipeline 😁

[kaf-lamed-beyt]

Oh! I'm not so sure. But, the last time I did a bucket setup was with tigris — I think it was with the guide you shared with me a while ago.

And thanks for letting me know the scale of the issue 😅.

I had already gone to read about transmuxing. From the issue description, I think it's a bit related, no?

[lastcanal]

@kaf-lamed-beyt your S3 bucket is not properly configured. Feel free to reach out on Discord if you have questions about getting setup.

@zcesur I added crash_group_mode: :transient to the group with the crashing muxer in SinkBin and that gave me this very helpful error message:

[error] <0.785.0>/ Error occured in Membrane Pipeline:
** (Membrane.CallbackError) Bin :sink removed its pad {Membrane.Pad, :input, "video_master"}, but callback `c:Membrane.Pipeline.handle_child_pad_removed/4` is not implemented in Algora.Pipeline.

This means, that `:sink removed the pad on its own, without knowledge of its parent. For example it could happen due to removing :sink's child linked to the :sink's inner pad or by removing link between :sink and its child.

If you want to handle this scenario, implement `c:Membrane.Pipeline.handle_child_pad_removed/4` callback in Algora.Pipeline.

Might be possible to recover by adding Algora.Pipeline.handle_child_pad_removed/4, which will allow us to re-initialize the HLS tracks, the HLS component itself, or at least do an orderly shutdown of the pipeline.