[Перевод] Ваш стриминг ненастоящий. Ловушка загрузки файлов в Spring Boot

Введение

Вы наверняка видели такой пример: обработчик с @PostMapping, параметр MultipartFile, вызов transferTo(). Пять минут — и готово. На локальной машине всё работает отлично.

А потом в рабочую среду прилетает первый файл на 500 МБ, и под падает с OutOfMemoryError.

Проблема не в Вашем коде, а в допущении, на котором построено большинство руководств: файлы достаточно малы, чтобы целиком сохранить их в памяти или на диске ещё до запуска обработчика. В реальных системах — медиаплатформах, системах управления документами, здравоохранении, финтехе — это допущение быстро перестаёт работать.

Причём версия примера на WebFlux зачастую ничем не лучше. Просто буферизация происходит в менее очевидном месте.

Это первая часть небольшой серии о создании конвейера загрузки файлов на Java, готового к промышленной эксплуатации. Здесь мы разберём:

• Почему MultipartFile буферизует файл целиком ещё до того, как Вы получаете к нему доступ.

• Почему @RequestPart в WebFlux делает то же самое, в том числе в примере, который все копируют.

• Какой API WebFlux действительно обеспечивает потоковую обработку и как правильно им пользоваться.

Во второй части займёмся сохранением потока в S3, определением типа файла по нескольким первым байтам и безопасной проверкой на вирусы при больших нагрузках. Также разберём сценарий сбоя, о котором стоит узнать до публикации обработчика загрузки в интернете: безобидный на вид файл может вывести из строя весь сервер. Затем повторим всё это на сервлетном стеке для команд, которые не могут перейти на WebFlux. Поэтому, если Вы используете Spring MVC, оставайтесь с нами.

Почему MultipartFile — ловушка

Spring MultipartFile и его аналоги кажутся удобными, но под капотом сервлетный контейнер буферизует всё тело запроса — в памяти или во временном файле — ещё до вызова метода контроллера.

При загрузке видео размером 2 ГБ это означает:

• 2 ГБ занятой кучи или диска на каждую одновременную загрузку.

• Передача в хранилище начинается только после получения всего файла.

• Ваши собственные проверки не запускаются, пока система не выполнит весь объём операций ввода-вывода.

Важно понимать, какой именно вариант используется: сбои у них разные.

file-size-threshold: 0 — значение по умолчанию для Tomcat — приводит к сохранению всего тела во временный файл. Десять одновременных загрузок по 2 ГБ занимают 20 ГБ временного дискового пространства, которые Вы не планировали расходовать. При этом в целевое хранилище ничего не поступает до получения последнего байта.

Если порог выше размера файла, данные остаются в куче. Именно этот вариант убивает JVM, а вместе с ней — и все остальные обрабатываемые запросы.

Прежде чем двигаться дальше, сделаем важную оговорку: контейнер может отклонить запрос заранее на основании собственных ограничений. Например, Tomcat проверяет max-file-size во время разбора. Вы теряете не любую возможность отклонить запрос, а возможность сделать это по своим правилам: проверить квоту, применить политику допустимых типов содержимого или начать сохранение до того, как последний байт попадёт во временное хранилище. Во второй части разберём, как выполнять эти проверки с небольшими затратами.

Общая картина: к чему мы движемся

Прежде чем погружаться в особенности WebFlux, посмотрим на весь конвейер, который построим в этой серии. Так будет понятнее, какое место занимает описанная ниже часть:

1. Принимаем загружаемый файл потоком — никогда не буферизуем его целиком (эта статья).

2. Читаем первые ~500 байт, чтобы определить тип файла, затем возвращаем прочитанные байты в поток (часть 2).

3. Если тип недопустим, сразу прекращаем обработку — без лишней записи в хранилище и проверки на вирусы.

4. Если тип допустим, направляем один входящий поток одновременно в:

     - хранилище (S3 или диск) — без ограничения объёма этой ветки: нужно сохранить весь файл;

     - антивирус — с разумным ограничением объёма, чтобы огромный файл не перегрузил сканер.

Общий принцип на каждом этапе: читаем входящие байты из сети ровно один раз и начинаем действовать, не дожидаясь, пока весь файл окажется в памяти или на диске. Эта статья целиком посвящена первому шагу — получению настоящего потока из входящего запроса. На этой основе во второй части построим сохранение и анализ.

В чём ошибается большинство руководств по WebFlux

Вероятно, Вы уже видели такой вариант. Он не обеспечивает потоковую обработку:

@PostMapping(value = "/upload", consumes = MediaType.MULTIPART_FORM_DATA_VALUE)
public Mono<UploadResult> upload(
        @RequestPart("file") Flux<DataBuffer> fileParts,
        @RequestPart("metadata") UploadMetadata metadata) {

    return uploadService.streamToStorage(fileParts, metadata);
}

Выглядит правильно. Flux<DataBuffer> — реактивный поток блоков данных. Сам тип говорит «поток», значит, всё должно работать потоково. Но это не так, и Spring прямо сообщает об этом в документации:

Для потокового разбора multipart-данных можно использовать Flux<PartEvent>, возвращаемый PartEventHttpMessageReader, вместо @RequestPart. Использование @RequestPart предполагает доступ к отдельным частям по имени, как в Map, а значит, требует полного разбора multipart-данных.

В этом и заключается вся хитрость: @RequestPart означает «найди эту часть по имени», а искать части по имени в ещё не прочитанном теле нельзя. Поэтому Spring сначала читает всё.

Вот что происходит с каждой из сигнатур, к которым Вы могли бы обратиться:

Три из четырёх вариантов заслуживают отдельного рассмотрения, потому что каждый приводит к своим проблемам.

FilePart — это MultipartFile с временным файлом под капотом. Именно этот вариант часто вводит в заблуждение: FilePart выглядит современным реактивным решением. Но к моменту вызова обработчика часть уже полностью получена. Если её размер меньше maxInMemorySize, она находится в памяти, иначе — во временном файле .multipart на диске. Справочник настроек Spring Boot прямо описывает каталог временных файлов как «каталог для хранения файловых частей, размер которых превышает maxInMemorySize; игнорируется при использовании потоковой обработки через PartEvent». Такая настройка существует именно потому, что @RequestPart записывает части на диск до того, как Вы получаете к ним доступ.

Вариант с Flux<DataBuffer> падает громко и непонятно. Для отдельного DataBuffer нет реактивного адаптера, поэтому декодер переходит к обработке одиночного значения и объединяет Ваш «поток» в один составной буфер, ограниченный spring.http.codecs.max-in-memory-size. При стандартном лимите 256 КБ загрузка файла на 1 МБ возвращает HTTP 413 — чего никто не ожидает от обработчика, для которого не настраивали ограничение размера. Если увеличить лимит, чтобы «исправить» проблему, каждая загрузка будет целиком храниться в куче. Получите тот же OOM, только с дополнительными шагами.

DataBufferUtils.join() тоже не помогает. Обычно предлагают такой «практичный переходник»:

InputStream inputStream = DataBufferUtils.join(fileParts)
    .map(buf -> buf.asInputStream(true))
    .block();

Здесь две проблемы. join() собирает все буферы в один составной буфер, поэтому потребление памяти растёт вместе с размером файла. Именно этого мы и пытаемся избежать. А .block() выполняется в потоке цикла событий Netty, в котором был вызван обработчик, поэтому до этого дело даже не доходит:

java.lang.IllegalStateException: block()/blockFirst()/blockLast() are blocking,
which is not supported in thread reactor-http-nio-4

Это не вопрос настройки производительности. Такой код просто не может выполняться.

Spring Boot: правильная потоковая обработка multipart в WebFlux

API, который действительно обеспечивает потоковую обработку, — @RequestBody Flux<PartEvent>. Вместо набора именованных частей Вы получаете плоский поток событий по мере разбора тела запроса. Поле формы создаёт один FormPartEvent, а файл — один или несколько FilePartEvent, каждый из которых содержит буфер. У последнего события каждой части isLast() возвращает true. Это позволяет разделять поток на части без буферизации.

@RestController
public class UploadController {

    private final PartEventUploadService uploads;

    @PostMapping(value = "/api/upload", consumes = MediaType.MULTIPART_FORM_DATA_VALUE)
    public Mono<ResponseEntity<UploadResult>> upload(@RequestBody Flux<PartEvent> events) {
        return uploads.store(events)
                .map(result -> ResponseEntity.status(HttpStatus.CREATED).body(result));
    }
}

Сервис обрабатывает части по одной, в порядке поступления:

public Mono<UploadResult> store(Flux<PartEvent> events) {
    AtomicReference<UploadMetadata> metadata = new AtomicReference<>();

    return events
            .windowUntil(PartEvent::isLast)                    // one window per part
            .concatMap(window -> handlePart(window, metadata)) // strictly sequential
            .next()
            .switchIfEmpty(Mono.error(new InvalidUploadException("No 'file' part in the request")));
}

Здесь принципиальны две детали.

windowUntil(PartEvent::isLast) разбивает плоский поток событий на окна, соответствующие отдельным частям. Затем concatMap обрабатывает их по очереди, поэтому телу запроса негде накапливаться. А поскольку проход всего один, порядок частей имеет значение: метаданные нужно отправлять перед файлом. Второго прохода, во время которого можно было бы их найти, не будет. Это реальное ограничение, а не вопрос стиля — цена отказа от буферизации.

Вторая деталь приводит к утечкам памяти, если сделать всё неправильно. Буферы содержимого FilePartEvent необходимо потребить, передать дальше или освободить — каждый без исключения. Фреймворк не сделает это за Вас.

Потоковая запись в хранилище

Модуль хранения принимает буферы файловой части и записывает их по мере поступления:

@Override
public Mono<StoredUpload> store(Flux<DataBuffer> content, StorageSpec spec) {
    Path tempFile = root.resolve(key + ".part");
    Path targetFile = root.resolve(key);
    AtomicLong written = new AtomicLong();
    MessageDigest digest = sha256();

    Flux<DataBuffer> metered = content
            .doOnDiscard(DataBuffer.class, DataBufferUtils::release)
            .doOnNext(buffer -> {
                long total = written.addAndGet(buffer.readableByteCount());
                if (total > maxBytes) {
                    DataBufferUtils.release(buffer);          // we are aborting: this one is ours to free
                    throw new UploadTooLargeException(maxBytes, total);
                }
                digest.update(buffer.asByteBuffer().duplicate());
            })
            .map(FileSystemStorageBackend::toHeapBuffer);

    return Mono.usingWhen(
            openChannel(tempFile),
            channel -> DataBufferUtils.write(metered, channel)
                    // write() re-emits each buffer once it is on disk: releasing is our job
                    .doOnNext(DataBufferUtils::release)
                    .doOnDiscard(DataBuffer.class, DataBufferUtils::release)
                    .then(Mono.fromCallable(() -> commit(channel, tempFile, targetFile, digest, written))),
            channel -> closeQuietly(channel),
            (channel, cause) -> closeQuietly(channel).then(deleteQuietly(tempFile)).then(Mono.error(cause)),
            channel -> closeQuietly(channel).then(deleteQuietly(tempFile)));
}

Здесь стоит отдельно отметить четыре момента: в наивной реализации каждый из них становится ошибкой.

DataBufferUtils.write не освобождает записанные буферы — он повторно выдаёт их дальше. Освобождать их должен подписчик. Пропустите эту строку — и при загрузках будет постоянно утекать память из пула за пределами кучи, пока процесс не завершится. Это самая распространённая ошибка в самописном коде потоковой обработки.

При ошибке потока уже записанные байты остаются на диске. Частичный файл удаляется при ошибке и отмене, например при отключении клиента. Итоговый файл появляется только через атомарное переименование, поэтому недописанная загрузка никогда не становится доступна по ключу, выданному вызывающей стороне.

Ограничение размера проверяется прямо во время передачи. Поэтому слишком большой файл отклоняется до завершения загрузки. Но есть компромисс, о котором нужно знать: при прерывании загрузки клиент может увидеть разрыв соединения раньше, чем получит тело ответа с ошибкой. Выполняйте дешёвые проверки до начала передачи тела. Если вызывающей стороне нужен корректный ответ, сначала дочитайте входящий поток, а затем возвращайте ошибку.

Перед записью каждый блок копируется в кучу, потому что передача прямых буферов из пула Netty непосредственно файловому каналу работает не на всех платформах. В Windows с актуальной JDK это приводит к исключению внутри JDK:

UnsupportedOperationException: ByteBuffer derived from closeable shared sessions not supported. 

Каналу нужен стабильный адрес, которого у такого буфера нет. Один memcpy на блок — и код работает везде.

Механизм обратного давления, или backpressure, получается автоматически, и полезно понимать, как он устроен. DataBufferUtils.write запрашивает следующий буфер только после завершения предыдущей записи. Этот спрос передаётся через парсер обратно к сокету. Медленный диск замедляет клиента, а не заполняет Вашу кучу.

Как выглядит запрос curl

Поскольку части обрабатываются по порядку, а JSON-часть с явно указанным типом содержимого не считается для Spring «полем формы», запрос выглядит так:

curl -X POST http://localhost:8080/api/upload \
  -F 'metadata={"filename":"big.bin","contentType":"application/octet-stream"}' \
  -F 'file=@big.bin'

Сначала метаданные в обычном поле. Затем файл. Обработчик в репозитории к статье принимает и вариант с явно указанным типом содержимого, но предпочтителен обычный вариант поля.

Действительно ли это потоковая обработка? Измеряем вместо утверждений

Утверждения вроде «потребление памяти зависит от размера блока, а не от размера файла» требуют доказательств. Поэтому в репозитории к статье есть два теста, которые я хотел бы увидеть, прежде чем поверить подобным заявлениям.

Байты попадают на диск до завершения загрузки. Тестовый клиент отправляет один блок и отказывается отправлять остальные, пока не обнаружит на сервере непустой файл .part. Реализация с буферизацией не сможет пройти этот тест: до получения всего тела она ничего не запишет, а клиент будет ждать бесконечно. Поэтому тест завершится по тайм-ауту, а не пройдёт по случайной причине. При потоковой обработке тест обнаруживает на диске около 59 КБ, пока оставшиеся 4 МБ ещё не отправлены.

Файл в несколько раз больше кучи. Приложение запускается отдельным процессом с жёстким ограничением -Xmx256m, а тест загружает 512 МБ, генерируемых на лету, — без огромного временного файла. Затем проверяет SHA-256 данных, записанных на диск:

Те же байты, та же куча — меняется только обработчик. Вся аргументация в одной таблице.

Две мои первоначальные ошибки

Обе обнаружились благодаря измерениям. Обе сохранились в истории репозитория, а не были исправлены незаметно.

На пропускную способность влияет размер блока, а не группировка записей. По умолчанию HTTP-декодер Reactor Netty передаёт приложению данные порциями по 8 КБ, поэтому загрузка на 512 МБ поступает примерно в виде 65 тысяч отдельных событий частей. Увеличение maxChunkSize декодера до 256 КБ сократило их количество примерно до 8200, а время той же загрузки — с 29 до 14,5 секунды. При этом характер потребления памяти совершенно не изменился: буферы по-прежнему обрабатываются по одному. Для ориентира: скорость обычной записи на диск на той же машине составляет около 81 МБ/с, поэтому на этом этапе ограничением становятся диск и вычисление хеша, а не сама потоковая архитектура.

Более хитрый оператор группировки тоже оказался ошибкой. Объединение нескольких блоков для одной записи изменило время менее чем на 5% и оказалось небезопасным: bufferTimeout выдаёт группы по таймеру независимо от спроса следующего обработчика, тогда как запись запрашивается строго по одной операции за раз. Под нагрузкой это приводило к OverflowException: Could not emit buffer due to lack of requests. Набор тестов не выявил проблему, поскольку она зависит от временных соотношений; её обнаружил ручной цикл запросов curl. Оператор удалили, и оставшийся путь полностью управляется спросом. Именно это и обеспечивает настоящее обратное давление.

Несколько слов об ограничениях

Если из этой статьи Вы вынесете только один совет по настройке, пусть это будет следующий: в WebFlux max-in-memory-size не определяет максимальный размер принимаемого файла.

• spring.webflux.multipart.max-in-memory-size — для @RequestPart задаёт порог, после которого часть переносится на диск. Для PartEvent ограничивает только поля формы. Размер файла эта настройка не ограничивает.

• spring.webflux.multipart.max-disk-usage-per-part — реальный предел размера потоково обрабатываемой части, который проверяется ещё во время поступления байтов.

• spring.http.codecs.max-in-memory-size — ограничивает декодирование с предварительным накоплением данных, включая рассмотренный выше случай @RequestPart Flux<DataBuffer>.

• Ваше собственное ограничение, проверяемое по мере поступления байтов, — позволяет вернуть понятную ошибку вместо стандартного отказа фреймворка.

Если Вы используете Boot 4, учтите: spring.codec.max-in-memory-size больше не существует. Теперь настройка называется spring.http.codecs.max-in-memory-size, и все руководства со старым именем незаметно для Вас ничего не настраивают.

Что дальше

Далее разберём реализацию всего описанного на сервлетном стеке: request.getInputStream(), исключение обработчика загрузки из работы multipart resolver, буферы ограниченного размера и политику пула потоков, которая не даст быстрому клиенту исчерпать ресурсы при работе с медленным диском. Если Вы используете Spring MVC и всё это время думали: «К моему приложению это не относится», — следующая часть для Вас. Чтобы отказаться от буферизации загружаемых файлов, переходить на WebFlux необязательно.

После этого займёмся конвейером, в который поступают загруженные данные:

• Хранение: потоковая передача тех же байтов в S3 через multipart upload, по-прежнему без буферизации, с использованием AsyncRequestBody.fromPublisher() вместо возврата к переходнику через InputStream.

• Определение типа: чтение примерно 500 байт вместо всего файла, чтобы решить, принимать ли его вообще; почему Apache Tika в качестве встроенной библиотеки создаёт риск при обработке вредоносных входных данных и как эту проблему решает tika-server.

• Антивирус: ограничение объёма проверки ClamAV, чтобы загрузка на 100 ГБ не перегрузила сканер, и обработка файлов, которые слишком велики для полной проверки.

Заключение

Наивная загрузка файлов работает, пока размер файлов или количество одновременных запросов не превышают возможности буферизации. В WebFlux этот подход коварнее: @RequestPart и FilePart выглядят как потоковые API, хотя сначала незаметно разбирают всё тело целиком. FilePart — это MultipartFile с временным файлом, переодетый в реактивную оболочку. По-настоящему потоковый API — @RequestBody Flux<PartEvent>. Но он требует от Вас ответных усилий: обрабатывать части по порядку, освобождать каждый буфер, с которым Вы работаете, удалять незавершённые файлы при обрыве передачи и проверять собственные ограничения прямо во время поступления данных.

Если всё сделать правильно, загрузка на 512 МБ будет занимать столько же памяти в куче, сколько загрузка на 5 МБ. Это основа, на которой строится всё остальное в этой серии.

Полный исходный код, включая тесты, подтверждающие каждое утверждение статьи, доступен на GitHub: denis111/spring-boot-large-file-upload.