Handler stuck when receiving request from ffmpeg

[marcoNjoHarmonic]

• I have looked for existing issues (including closed) about this

Bug Report

Version

├── axum v0.6.20
│   ├── axum-core v0.3.4
│   ├── axum-macros v0.3.8 (proc-macro)

Platform

Linux 0bd88cd80067 6.8.0-39-generic #39-Ubuntu SMP PREEMPT_DYNAMIC Sat Jul  6 02:50:39 UTC 2024 aarch64 aarch64 aarch64 GNU/Linux

Description

I tried upload files (.m3u8 and .m4v) to the following code with ffmpeg 4.2.10

mod app_settings;
mod app_state;
mod database;
mod errors;
mod event;
mod http_handlers;
mod jwt;
mod metrics;
mod models;
mod processes;
mod stream;
mod utilities;

use axum::extract::{BodyStream, Path};
use axum::routing::put;
use futures::StreamExt;
use hyper::StatusCode;

use crate::app_settings::AppSettings;

use axum::{Router, Server};

use foundations::{
    cli::{Arg, ArgAction, Cli},
    telemetry::{init_with_server, log},
    BootstrapResult,
};

#[tokio::main]
async fn main() -> BootstrapResult<()> {
    let service_info = foundations::service_info!();

    let cli = Cli::<AppSettings>::new(
        &service_info,
        vec![Arg::new("dry-run")
            .long("dry-run")
            .action(ArgAction::SetTrue)
            .help("Validate or generate config without running the server")],
    )?;

    if cli.arg_matches.get_flag("dry-run") {
        return Ok(());
    }

    let tele_serv_fut = init_with_server(&service_info, &cli.settings.telemetry, vec![])?;

    if let Some(tele_serv_addr) = tele_serv_fut.server_addr() {
        log::info!("Telemetry server is listening on http://{}", tele_serv_addr);
    }

    let router = Router::new().route("/*filepath", put(upload_file));

    let server = Server::bind(&cli.settings.bind).serve(router.into_make_service());
    log::info!("Listening on http://{}", server.local_addr());

    tokio::select! {
        _ = server => {
            log::error!("Server has stopped unexpectedly");
        }
        _ = tele_serv_fut => {
            log::error!("Telemetry server has stopped unexpectedly");
        }
    }
    return Ok(());
}

async fn upload_file(
    Path((stream_id, event_name, filepath)): Path<(String, String, String)>,
    mut body: BodyStream,
) -> Result<StatusCode, StatusCode> {
    let mut aggregated_body = Vec::new();
    while let Some(chunk) = body.next().await {
        let chunk = chunk.unwrap();
        log::info!("{}, Received chunk of size: {}", filepath, chunk.len());
        aggregated_body.extend_from_slice(&chunk);
    }

    log::info!(
        "{}, Aggregated file of size: {}",
        filepath,
        aggregated_body.len()
    );
    Ok(StatusCode::OK)
}

ffmpeg command

exec ffmpeg -re -loop 1 -i "$SCRIPT_DIR/sample-stream-sample.jpg" -preset ultrafast \
-vf "drawtext=text='%{localtime\:%H}\\:%{localtime\:%M}\\:%{localtime\:%S}':fontsize=180:fontcolor=white:x=(w-text_w)/2:y=(h-text_h)/2,drawtext=text='$TEXT':fontsize=180:fontcolor=white:x=(w-text_w)/2:y=(h-text_h)/4" \
-r 10 -c:v libx264 -pix_fmt yuv420p -x264-params 'keyint=10:nal-hrd=cbr' \
-hls_time 4 -hls_list_size 100 -hls_flags delete_segments+program_date_time \
-hls_segment_type fmp4 -hls_segment_filename "$ENDPOINT/segment_%v_%03d.m4v" -master_pl_name "index.m3u8" \
-map v:0 -s:v:0 1280x720 -b:v:0 3000k -minrate:v:0 3000k -maxrate:v:0 3000k -bufsize:v:0 3000k \
-map v:0 -s:v:1 854x480 -b:v:1 1000k -minrate:v:0 1000k -maxrate:v:1 1000k -bufsize:v:1 1000k \
-var_stream_map "v:0,name:0 v:1,name:1" -http_persistent 1 -f hls "$ENDPOINT/stream_%v.m3u8"

I expect to see server logs containing

Oct 10 23:18:33.107 INFO segment_1_004.m4v, Received chunk of size: 32768, pid: 3289902, version: 0.1.0
Oct 10 23:18:33.107 INFO segment_1_004.m4v, Received chunk of size: 32492, pid: 3289902, version: 0.1.0
Oct 10 23:18:33.148 INFO segment_1_004.m4v, Received chunk of size:<some size> , pid: 3289902, version: 0.1.0

But I got the following only, it appears the code stuck in body.next().await

Oct 10 23:18:33.107 INFO segment_1_004.m4v, Received chunk of size: 32768, pid: 3289902, version: 0.1.0
Oct 10 23:18:33.107 INFO segment_1_004.m4v, Received chunk of size: 32492, pid: 3289902, version: 0.1.0

I have tried changing the extractor from BodyStream to Byte, but in that case, I did not get any of the above-provided logs.
I have also tried to upload files with the same ffmpeg version and command to a Java server and did not suffer from this issue.

[ttys3]

better provide a whole project and link to the video, so people could reproduce your problem

[mladedav]

I believe ffmpeg is half closing the connection. You need to use some hyper settings to allow that. Either see the hyper examples in this repo or try using axum_server.

[marcoNjoHarmonic]

@mladedav Your guess is correct, thanks a lot!