Clean multipart up only when data is read

This commit ensures that any storage used for multipart handling only
gets cleaned up if multipart data is actually retrieved via
ServerWebExchange::getMultipartData.

Closes gh-30590

Author: poutsma

Diff for: spring-web/src/main/java/org/springframework/web/server/ServerWebExchange.java

@@ -1,5 +1,5 @@
 /*
- * Copyright 2002-2019 the original author or authors.
+ * Copyright 2002-2023 the original author or authors.
  *
  * Licensed under the Apache License, Version 2.0 (the "License");
  * you may not use this file except in compliance with the License.
@@ -141,6 +141,13 @@ default <T> T getAttributeOrDefault(String name, T defaultValue) {
 	 */
 	Mono<MultiValueMap<String, Part>> getMultipartData();
 
+	/**
+	 * Cleans up any storage used for multipart handling.
+	 * @since 6.0.10
+	 * @see Part#delete()
+	 */
+	Mono<Void> cleanupMultipart();
+
 	/**
 	 * Return the {@link LocaleContext} using the configured
 	 * {@link org.springframework.web.server.i18n.LocaleContextResolver}.

Diff for: spring-web/src/main/java/org/springframework/web/server/ServerWebExchangeDecorator.java

@@ -1,5 +1,5 @@
 /*
- * Copyright 2002-2019 the original author or authors.
+ * Copyright 2002-2023 the original author or authors.
  *
  * Licensed under the Apache License, Version 2.0 (the "License");
  * you may not use this file except in compliance with the License.
@@ -108,6 +108,11 @@ public Mono<MultiValueMap<String, Part>> getMultipartData() {
 		return getDelegate().getMultipartData();
 	}
 
+	@Override
+	public Mono<Void> cleanupMultipart() {
+		return getDelegate().cleanupMultipart();
+	}
+
 	@Override
 	public boolean isNotModified() {
 		return getDelegate().isNotModified();

Diff for: spring-web/src/main/java/org/springframework/web/server/adapter/DefaultServerWebExchange.java

@@ -93,6 +93,8 @@ public class DefaultServerWebExchange implements ServerWebExchange {
 
 	private final Mono<MultiValueMap<String, Part>> multipartDataMono;
 
+	private volatile boolean multipartRead = false;
+
 	@Nullable
 	private final ApplicationContext applicationContext;
 
@@ -131,7 +133,7 @@ public DefaultServerWebExchange(ServerHttpRequest request, ServerHttpResponse re
 		this.sessionMono = sessionManager.getSession(this).cache();
 		this.localeContextResolver = localeContextResolver;
 		this.formDataMono = initFormData(request, codecConfigurer, getLogPrefix());
-		this.multipartDataMono = initMultipartData(request, codecConfigurer, getLogPrefix());
+		this.multipartDataMono = initMultipartData(codecConfigurer, getLogPrefix());
 		this.applicationContext = applicationContext;
 	}
 
@@ -154,10 +156,9 @@ private static Mono<MultiValueMap<String, String>> initFormData(ServerHttpReques
 				.cache();
 	}
 
-	private static Mono<MultiValueMap<String, Part>> initMultipartData(ServerHttpRequest request,
-			ServerCodecConfigurer configurer, String logPrefix) {
+	private Mono<MultiValueMap<String, Part>> initMultipartData(ServerCodecConfigurer configurer, String logPrefix) {
 
-		MediaType contentType = getContentType(request);
+		MediaType contentType = getContentType(this.request);
 		if (contentType == null || !contentType.getType().equalsIgnoreCase("multipart")) {
 			return EMPTY_MULTIPART_DATA;
 		}
@@ -168,7 +169,8 @@ private static Mono<MultiValueMap<String, Part>> initMultipartData(ServerHttpReq
 		}
 
 		return reader
-				.readMono(MULTIPART_DATA_TYPE, request, Hints.from(Hints.LOG_PREFIX_HINT, logPrefix))
+				.readMono(MULTIPART_DATA_TYPE, this.request, Hints.from(Hints.LOG_PREFIX_HINT, logPrefix))
+				.doOnNext(ignored -> this.multipartRead = true)
 				.switchIfEmpty(EMPTY_MULTIPART_DATA)
 				.cache();
 	}
@@ -243,6 +245,22 @@ public Mono<MultiValueMap<String, Part>> getMultipartData() {
 		return this.multipartDataMono;
 	}
 
+	@Override
+	public Mono<Void> cleanupMultipart() {
+		if (this.multipartRead) {
+			return getMultipartData()
+					.onErrorResume(t -> Mono.empty()) // ignore errors reading multipart data
+					.flatMapIterable(Map::values)
+					.flatMapIterable(Function.identity())
+					.flatMap(part -> part.delete()
+									.onErrorResume(ex -> Mono.empty()))
+					.then();
+		}
+		else {
+			return Mono.empty();
+		}
+	}
+
 	@Override
 	public LocaleContext getLocaleContext() {
 		return this.localeContextResolver.resolveLocaleContext(this);

Diff for: spring-web/src/main/java/org/springframework/web/server/adapter/HttpWebHandlerAdapter.java

@@ -1,5 +1,5 @@
 /*
- * Copyright 2002-2022 the original author or authors.
+ * Copyright 2002-2023 the original author or authors.
  *
  * Licensed under the Apache License, Version 2.0 (the "License");
  * you may not use this file except in compliance with the License.
@@ -16,9 +16,7 @@
 
 package org.springframework.web.server.adapter;
 
-import java.util.Map;
 import java.util.Set;
-import java.util.function.Function;
 
 import org.apache.commons.logging.Log;
 import org.apache.commons.logging.LogFactory;
@@ -32,7 +30,6 @@
 import org.springframework.http.HttpStatusCode;
 import org.springframework.http.codec.LoggingCodecSupport;
 import org.springframework.http.codec.ServerCodecConfigurer;
-import org.springframework.http.codec.multipart.Part;
 import org.springframework.http.server.reactive.HttpHandler;
 import org.springframework.http.server.reactive.ServerHttpRequest;
 import org.springframework.http.server.reactive.ServerHttpResponse;
@@ -250,7 +247,7 @@ public Mono<Void> handle(ServerHttpRequest request, ServerHttpResponse response)
 		return getDelegate().handle(exchange)
 				.doOnSuccess(aVoid -> logResponse(exchange))
 				.onErrorResume(ex -> handleUnresolvedError(exchange, ex))
-				.then(cleanupMultipart(exchange))
+				.then(exchange.cleanupMultipart())
 				.then(Mono.defer(response::setComplete));
 	}
 
@@ -325,23 +322,4 @@ private boolean isDisconnectedClientError(Throwable ex) {
 		return DISCONNECTED_CLIENT_EXCEPTIONS.contains(ex.getClass().getSimpleName());
 	}
 
-	private Mono<Void> cleanupMultipart(ServerWebExchange exchange) {
-		return exchange.getMultipartData()
-				.onErrorResume(t -> Mono.empty()) // ignore errors reading multipart data
-				.flatMapIterable(Map::values)
-				.flatMapIterable(Function.identity())
-				.flatMap(this::deletePart)
-				.then();
-	}
-
-	private Mono<Void> deletePart(Part part) {
-		return part.delete().onErrorResume(ex -> {
-					if (logger.isWarnEnabled()) {
-						logger.warn("Failed to perform cleanup of multipart items", ex);
-					}
-					return Mono.empty();
-				});
-	}
-
-
 }

Diff for: spring-webflux/src/main/java/org/springframework/web/reactive/function/server/DefaultServerRequestBuilder.java

@@ -324,28 +324,30 @@ private static class DelegatingServerWebExchange implements ServerWebExchange {
 
 		private final Mono<MultiValueMap<String, Part>> multipartDataMono;
 
+		private volatile boolean multipartRead = false;
+
+
 		DelegatingServerWebExchange(ServerHttpRequest request, Map<String, Object> attributes,
 				ServerWebExchange delegate, List<HttpMessageReader<?>> messageReaders) {
 
 			this.request = request;
 			this.attributes = attributes;
 			this.delegate = delegate;
-			this.formDataMono = initFormData(request, messageReaders);
+			this.formDataMono = initFormData(messageReaders);
 			this.multipartDataMono = initMultipartData(request, messageReaders);
 		}
 
 		@SuppressWarnings("unchecked")
-		private static Mono<MultiValueMap<String, String>> initFormData(ServerHttpRequest request,
-				List<HttpMessageReader<?>> readers) {
-
+		private Mono<MultiValueMap<String, String>> initFormData(List<HttpMessageReader<?>> readers) {
 			try {
-				MediaType contentType = request.getHeaders().getContentType();
+				MediaType contentType = this.request.getHeaders().getContentType();
 				if (MediaType.APPLICATION_FORM_URLENCODED.isCompatibleWith(contentType)) {
 					return ((HttpMessageReader<MultiValueMap<String, String>>) readers.stream()
 							.filter(reader -> reader.canRead(FORM_DATA_TYPE, MediaType.APPLICATION_FORM_URLENCODED))
 							.findFirst()
 							.orElseThrow(() -> new IllegalStateException("No form data HttpMessageReader.")))
-							.readMono(FORM_DATA_TYPE, request, Hints.none())
+							.readMono(FORM_DATA_TYPE, this.request, Hints.none())
+							.doOnNext(ignored -> this.multipartRead = true)
 							.switchIfEmpty(EMPTY_FORM_DATA)
 							.cache();
 				}
@@ -398,7 +400,23 @@ public Mono<MultiValueMap<String, Part>> getMultipartData() {
 			return this.multipartDataMono;
 		}
 
-		// Delegating methods
+		@Override
+		public Mono<Void> cleanupMultipart() {
+			if (this.multipartRead) {
+				return getMultipartData()
+						.onErrorResume(t -> Mono.empty()) // ignore errors reading multipart data
+						.flatMapIterable(Map::values)
+						.flatMapIterable(Function.identity())
+						.flatMap(part -> part.delete()
+										.onErrorResume(ex -> Mono.empty()))
+						.then();
+			}
+			else {
+				return Mono.empty();
+			}
+		}
+
+        // Delegating methods
 
 		@Override
 		public ServerHttpResponse getResponse() {