Repository: clj-commons/aleph
Pull request #656 (2023-01-08 20:00:35)

multipart: manual cleanup to handle temporary files
## Description

This is a followup PR for #596 (which has been closed in favor of this) to propose a less risky and a backwards compatible approach for multipart requests.

Using temporary files storage is broken with the current `release all resources on stream close` mechanism. Fixing it is barely feasible without introducing a breaking change. 
Still, the current mechanism is really convenient: you consume the stream and you are done! It's working perfectly fine as long as you keep all your data on memory. 
__=> Let's keep it that way.__

I propose here to introduce a `manual-cleanup?` flag (a better name can be proposed) which will return a vector composed of a manifold.stream and a `cleanup` function which can be called after the stream has been consumed.

This PR doesn't aim at bringing a file granularity clean up function (which I think is not necessary) as proposed by #596.

And there is also a cljdoc article attached to this PR : https://github.com/clj-commons/aleph/compare/multipart-requests?expand=1#diff-45f6713a5206dde4c1c69d8ddc5242e5ed09e8256ecec0ab80ff911bb8e967c3R1-R82

Closes : #567

arnaudgeiser: I had to rebase on master (and handle merge conflicts), but your comments have been addressed.
I also bumped to Clojure 1.11 to have access to `parse-uuid`.
arnaudgeiser: Final rebase and merge.

Patch:
diff --git a/docs/cljdoc.edn b/docs/cljdoc.edn
index 05c1369..c29a31e 100644
--- a/docs/cljdoc.edn
+++ b/docs/cljdoc.edn
@@ -1,4 +1,5 @@
 {:cljdoc.doc/tree [["Read Me" {:file "README.md"}]
-                   ["HTTP" {:file "docs/http.md"}]
+                   ["HTTP" {:file "docs/http.md"}
+                    ["Handling multipart requests" {:file "docs/http/multipart.md"}]]
                    ["TCP" {:file "docs/tcp.md"}]
                    ["UDP" {:file "docs/udp.md"}]]}
diff --git a/docs/http/multipart.md b/docs/http/multipart.md
new file mode 100644
index 0000000..38f5290
--- /dev/null
+++ b/docs/http/multipart.md
@@ -0,0 +1,83 @@
+# Handling multipart requests
+
+## Defining the handler
+
+Aleph allows you to handle multipart requests through the usage of the `aleph.http.multipart/decode-request`
+function. This function returns a `manifold.stream` of parts having the following
+specification :
+
+```clojure
+(s/def ::part-name         string?)
+(s/def ::content           (s/or :string       string?
+                                 :input-stream (partial instance? java.io.InputStream)))
+(s/def ::name              (s/nilable string?))
+(s/def ::charset           string?)
+(s/def ::mime-type         string?)
+(s/def ::transfer-encoding (s/nilable string?))
+(s/def ::memory?           boolean?)
+(s/def ::file?             boolean?)
+(s/def ::file              (s/nilable (partial instance? java.io.File)))
+(s/def ::size              int?)
+
+(s/def ::part (s/keys
+               :req-un [::part-name
+                        ::content
+                        ::name
+                        ::charset
+                        ::mime-type
+                        ::transfer-encoding
+                        ::memory?
+                        ::file?
+                        ::file
+                        ::size]))
+```
+
+Every file can be stored either on memory or on the filesystem. It depends on the
+`memory-limit` passed to `decode-request` which defaults to 16 KB.
+
+By default, all the resources are automatically cleaned up as soon as the stream is closed.
+It means that all the temporary files created on the filesystem will be removed.
+
+```clojure
+(require '[aleph.http.multipart :as mp])
+(require '[clojure.java.io      :as io])
+(require '[manifold.stream      :as s])
+
+(defn multipart-handler [req]
+  (let [s (mp/decode-request req {:memory-limit (* 16 1024)})]
+    (doseq [part (s/stream->seq s)]
+      (if (:memory? part))
+        (:content part)                           ;; do what you want with the content
+        (io/copy (:file part) (io/file "..."))))) ;; copy the file on another location
+```
+
+## Limitations
+
+As the consumption of the stream does not happen on the same Thread that cleans the
+resources, the files might be removed before you have the time to copy them on
+another location.
+If you want to rely on that convenient mechanism, you need to ensure all the parts
+you want to handle will fit on memory by overriding the default value.
+Depending on your workload, having a value between `8-100` MiB might be affordable.
+
+```clojure
+(mp/decode-request req {:memory-limit (* 16 1024 1024)})
+```
+
+## Manual cleanup
+
+If you cannot afford that much data on memory and you want to rely on temporary files, 
+you will have to disable the automatic clean up of the resources by passing the parameter
+`{:manual-cleanup? true}`.
+Instead of returning a manifold stream, it will return a vector composed
+of a manifold stream and a callback to clean the associated resources.
+
+```clojure
+(defn multipart-handler [req]
+  (let [[s cleanup-fn] (mp/decode-request req {:manual-cleanup? true})]
+    (doseq [part (s/stream->seq s)]
+      (if (:memory? part))
+         (:content part)         ;; do what you want with the content
+         (io/copy (:file part))) ;; copy the file on another location
+    (cleanup-fn)))
+```
diff --git a/project.clj b/project.clj
index 5278e80..c4a4690 100644
--- a/project.clj
+++ b/project.clj
@@ -25,7 +25,7 @@
                  [io.netty/netty-handler-proxy ~netty-version]
                  [io.netty/netty-resolver ~netty-version]
                  [io.netty/netty-resolver-dns ~netty-version]]
-  :profiles {:dev  {:dependencies [[org.clojure/clojure "1.10.3"]
+  :profiles {:dev  {:dependencies [[org.clojure/clojure "1.11.0"]
                                    [criterium "0.4.6"]
                                    [cheshire "5.10.0"]
                                    [org.slf4j/slf4j-simple "1.7.30"]
diff --git a/src/aleph/http/multipart.clj b/src/aleph/http/multipart.clj
index 1dc9af4..81c779d 100644
--- a/src/aleph/http/multipart.clj
+++ b/src/aleph/http/multipart.clj
@@ -227,9 +227,20 @@
      :file (when-not memory? (.getFile data))
      :size (.length data)}))
 
-(defn- read-attributes [^HttpPostRequestDecoder decoder parts]
+(defn- read-attributes [^HttpPostRequestDecoder decoder parts manual-cleanup?]
   (while (.hasNext decoder)
-    (s/put! parts (http-data->map (.next decoder)))))
+    (let [{:keys [file] :as part} (http-data->map (.next decoder))]
+      (when (and (not manual-cleanup?) file)
+        (log/warn (str "Temporary files storage is not working using automatic clean up. consider using manual cleanup.` "
+                       "See: https://cljdoc.org/d/aleph/aleph/CURRENT/doc/http/handling-multipart-requests")))
+      (s/put! parts part))))
+
+(defn- destroy-decoder [^HttpPostRequestDecoder decoder destroyed?]
+  (when (compare-and-set! destroyed? false true)
+    (try
+      (.destroy decoder)
+      (catch Exception e
+        (log/warn e "exception when cleaning up multipart decoder")))))
 
 (defn decode-request
   "Takes a ring request and returns a manifold stream which yields
@@ -239,23 +250,33 @@
    flag to know whether content might be read directly from `:content` or
    should be fetched from the file specified in `:file`.
 
+   If you want to use temporary files storage, you will have to fallback
+   to `:manual-cleanup?` otherwise the files will be removed from the filesystem
+   before you had time to copy them on another location.
+   Instead of returning a manifold stream, it will return vector composed of a
+   manifold stream and a callback to clean the resources.
+   See: https://cljdoc.org/d/aleph/aleph/CURRENT/doc/http/handling-multipart-requests#manual-cleanup
+
    Note, that if your handler works with multipart requests only,
    it's better to set `:raw-stream?` to `true` to avoid additional
    input stream coercion."
   ([req] (decode-request req {}))
   ([{:keys [body] :as req}
     {:keys [body-buffer-size
-            memory-limit]
+            memory-limit
+            manual-cleanup?]
      :or {body-buffer-size 65536
-          memory-limit DefaultHttpDataFactory/MINSIZE}}]
-   (let [body (if (s/stream? body)
-                body
-                (netty/to-byte-buf-stream body body-buffer-size))
-         destroyed? (atom false)
-         req' (http-core/ring-request->netty-request req)
-         factory (DefaultHttpDataFactory. (long memory-limit))
-         decoder (HttpPostRequestDecoder. factory req')
-         parts (s/stream)]
+          memory-limit DefaultHttpDataFactory/MINSIZE
+          manual-cleanup? false}}]
+   (let [body            (if (s/stream? body)
+                           body
+                           (netty/to-byte-buf-stream body body-buffer-size))
+         destroyed?      (atom false)
+         req'            (http-core/ring-request->netty-request req)
+         factory         (DefaultHttpDataFactory. (long memory-limit))
+         decoder         (HttpPostRequestDecoder. factory req')
+         parts           (s/stream)
+         destroy-decoder (partial destroy-decoder decoder destroyed?)]
 
      ;; on each HttpContent chunk, put it into the decoder
      ;; and resume our attempts to get the next attribute available
@@ -264,7 +285,7 @@
       (fn [chunk]
         (let [content (DefaultHttpContent. chunk)]
           (.offer decoder content)
-          (read-attributes decoder parts)
+          (read-attributes decoder parts manual-cleanup?)
           ;; note, that releasing chunk right here relies on
           ;; the internals of the decoder. in case those
           ;; internal are changed in future, this flow of
@@ -273,13 +294,8 @@
           (d/success-deferred true)))
       parts)
 
-     (s/on-closed
-      parts
-      (fn []
-        (when (compare-and-set! destroyed? false true)
-          (try
-            (.destroy decoder)
-            (catch Exception e
-              (log/warn e "exception when cleaning up multipart decoder"))))))
-
-     parts)))
+     (if manual-cleanup?
+       [parts destroy-decoder]
+       (do
+         (s/on-closed parts destroy-decoder)
+         parts)))))


Test patch:
diff --git a/test/aleph/http/multipart_test.clj b/test/aleph/http/multipart_test.clj
index 382b69d..41b9a58 100644
--- a/test/aleph/http/multipart_test.clj
+++ b/test/aleph/http/multipart_test.clj
@@ -5,6 +5,7 @@
    [aleph.http.multipart :as mp]
    [aleph.netty :as netty]
    [clj-commons.byte-streams :as bs]
+   [clojure.java.io :as io]
    [clojure.string :as str]
    [clojure.test :refer [deftest is testing]]
    [manifold.deferred :as d]
@@ -91,9 +92,9 @@
 #_{:clj-kondo/ignore [:deprecated-var]}
 (deftest reject-unknown-transfer-encoding
   (is (thrown? IllegalArgumentException
-       (mp/encode-body [{:part-name "part1"
-                         :content "content1"
-                         :transfer-encoding :uknown-transfer-encoding}]))))
+               (mp/encode-body [{:part-name "part1"
+                                 :content "content1"
+                                 :transfer-encoding :uknown-transfer-encoding}]))))
 
 #_{:clj-kondo/ignore [:deprecated-var]}
 (deftest test-content-as-file
@@ -191,55 +192,143 @@
 
     (.close ^java.io.Closeable s)))
 
-(defn- decode-handler [req]
-  (let [is     (:body req)
-        data   (bs/to-string is)
-        chunks (-> (assoc req :body (netty/to-byte-buf-stream (.getBytes data) 512))
-                   mp/decode-request
-                   s/stream->seq)]
-    {:status 200
-     :body (pr-str {:chunks (mapv #(update % :content bs/to-string) chunks)
-                    :encoded-data data})}))
-
-(defn- test-decoder [port url raw-stream?]
-  (let [s (http/start-server decode-handler {:port port
-                                             :shutdown-timeout 0
-                                             :raw-stream? raw-stream?})]
-    (try
-      (let [req          (http/post url {:multipart parts})
-            resp         (deref req 1e3 {:body "timeout"})
-            body         (-> (:body resp) bs/to-string read-string)
-            encoded-data (:encoded-data body)
-            chunks       (-> body :chunks vec)]
-        (is (= 7 (count chunks)))
-
-        ;; part-names
-        (is (= (map :part-name parts)
-               (map :part-name chunks)))
-
-        ;; content
-        (is (= "CONTENT1" (get-in chunks [0 :content])))
-
-        ;; mime type
-        (is (= "text/plain" (get-in chunks [2 :mime-type])))
-        (is (= "application/png" (get-in chunks [3 :mime-type])))
-
-        ;; filename
-        (is (= "file.txt" (get-in chunks [3 :name])))
-        (is (= "file.txt" (get-in chunks [4 :name])))
-
-        ;; charset
-        (is (= "ISO-8859-1" (get-in chunks [5 :charset])))
-
-        ;; mime-type + memory data
-        (is (re-find #"content-disposition: form-data; name=\"#6-bytes-with-mime-type\"; filename=\".*\"\r\ncontent-length: 8\r\ncontent-type: text/plain\r\ncontent-transfer-encoding: binary\r\n\r\nCONTENT3\r\n" encoded-data))
-        (is (= "CONTENT3" (get-in chunks [6 :content])))
-        (is (= "text/plain" (get-in chunks [6 :mime-type]))))
-
-      (finally (.close ^java.io.Closeable s)))))
+(defn- pack-part [{:keys [content file] :as part}]
+  (cond-> part
+    (some? file)
+    (assoc :file (.getAbsolutePath ^java.io.File file))
+    (some? content)
+    (update :content bs/to-string)))
+
+(defn- make-decode-handler [options]
+  (fn [req]
+    (let [body   (:body req)
+          data   (bs/to-string body)
+          parts  (-> (assoc req :body (netty/to-byte-buf-stream (.getBytes data) 512))
+                     (#(mp/decode-request % options))
+                     s/stream->seq)
+          packed-parts (mapv (fn [{:keys [file] :as part}]
+                               (Thread/sleep 20) ;; wait a bit to ensure the files are cleaned when using automatic resource cleanup
+                               (when file
+                                 (is (instance? java.io.File file))
+                                 (is (not (.exists ^java.io.File file)))) ;; demonstrates there is an issue when the stream is closed before accessing the file.
+                               (pack-part part))
+                             parts)]
+      {:status 200
+       :body (pr-str {:parts packed-parts
+                      :encoded-data data})})))
+
+(defn- make-decode-manual-handler [options]
+  (fn [req]
+    (let [body   (:body req)
+          data   (bs/to-string body)
+          [parts cleanup-fn]  (-> (assoc req :body (netty/to-byte-buf-stream (.getBytes data) 512))
+                                  (#(mp/decode-request % options)))
+          packed-parts (mapv (fn [{:keys [file] :as part}]
+                               (Thread/sleep 20) ;; wait a bit to ensure the files are cleaned when using automatic resource cleanup
+                               (when file
+                                 (is (some? (slurp (.getAbsolutePath ^java.io.File file)))))
+                               (pack-part part))
+                             (s/stream->seq parts))]
+      (is (every? #(.exists (io/file %)) (keep :file packed-parts)))
+      (cleanup-fn)
+      (is (every? #(not (.exists (io/file %))) (keep :file packed-parts)))
+      {:status 200
+       :body (pr-str {:parts packed-parts
+                      :encoded-data data})})))
+
+(defn- test-decoder
+  ([port url handler]
+   (test-decoder port url handler {}))
+  ([port url handler options]
+   (let [handler (bound-fn* handler)
+         s (http/start-server handler (merge {:port port
+                                              :shutdown-timeout 0}
+                                             options))]
+
+     (try
+       (let [req (http/post url {:multipart parts})
+             resp         (deref req 1e3 {:body "timeout"})
+             body         (-> (:body resp) bs/to-string read-string)
+             encoded-data (:encoded-data body)
+             parts-resp   (-> body :parts vec)]
+         (is (= 7 (count parts-resp)))
+
+         ;; part-names
+         (is (= (map :part-name parts)
+                (map :part-name parts-resp)))
+
+         ;; content
+         (is (= "CONTENT1" (get-in parts-resp [0 :content])))
+
+         ;; mime type
+         (is (= "text/plain" (get-in parts-resp [2 :mime-type])))
+         (is (= "application/png" (get-in parts-resp [3 :mime-type])))
+
+         ;; filename
+         (is (= "file.txt" (get-in parts-resp [3 :name])))
+         (is (= "file.txt" (get-in parts-resp [4 :name])))
+
+         ;; charset
+         (is (= "ISO-8859-1" (get-in parts-resp [5 :charset])))
+
+         ;; mime-type + memory data
+         (is (re-find #"content-disposition: form-data; name=\"#6-bytes-with-mime-type\"; filename=\".*\"\r\ncontent-length: 8\r\ncontent-type: text/plain\r\ncontent-transfer-encoding: binary\r\n\r\nCONTENT3\r\n" encoded-data))
+         (is (uuid? (parse-uuid (get-in parts-resp [6 :name]))))
+         (is (= "text/plain" (get-in parts-resp [6 :mime-type]))))
+
+       (finally (.close ^java.io.Closeable s))))))
 
 (deftest test-multipart-request-decode-with-ring-handler
-  (test-decoder port2 url2 false))
+  (testing "automatic cleanup"
+    (testing "without memory-limit"
+      (test-decoder port2 url2 (make-decode-handler {})))
+    (testing "with infinite memory-limit"
+      (test-decoder port2 url2 (make-decode-handler {:memory-limit Long/MAX_VALUE})))
+    (testing "with small memory-limit"
+      (test-decoder port2 url2 (make-decode-handler {:memory-limit 12})))
+    (testing "with zero memory-limit"
+      (test-decoder port2 url2 (make-decode-handler {:memory-limit 0}))))
+
+  (testing "manual cleanup"
+    (testing "without memory-limit"
+      (test-decoder port2 url2 (make-decode-manual-handler {:manual-cleanup? true})))
+    (testing "with infinite memory-limit"
+      (test-decoder port2 url2 (make-decode-manual-handler {:memory-limit Long/MAX_VALUE
+                                                            :manual-cleanup? true})))
+    (testing "with small memory-limit"
+      (test-decoder port2 url2 (make-decode-manual-handler {:memory-limit 12
+                                                            :manual-cleanup? true})))
+    (testing "with zero memory-limit"
+      (test-decoder port2 url2 (make-decode-manual-handler {:memory-limit 0
+                                                            :manual-cleanup? true})))))
 
 (deftest test-multipart-request-decode-with-raw-handler
-  (test-decoder port3 url3 true))
+  (testing "automatic cleanup"
+    (testing "without memory-limit"
+      (test-decoder port2 url2 (make-decode-handler {:raw-stream? true})))
+    (testing "with infinite memory-limit"
+      (test-decoder port2 url2 (make-decode-handler {:raw-stream? true
+                                                     :memory-limit Long/MAX_VALUE})))
+    (testing "with small memory-limit"
+      (test-decoder port2 url2 (make-decode-handler {:raw-stream? true
+                                                     :memory-limit 12})))
+    (testing "with zero memory-limit"
+      (test-decoder port2 url2 (make-decode-handler {:raw-stream? true
+                                                     :memory-limit 0}))))
+
+  (testing "manual cleanup"
+    (testing "without memory-limit"
+      (test-decoder port2 url2 (make-decode-manual-handler {:raw-stream? true
+                                                            :manual-cleanup? true})))
+    (testing "with infinite memory-limit"
+      (test-decoder port2 url2 (make-decode-manual-handler {:raw-stream? true
+                                                            :memory-limit Long/MAX_VALUE
+                                                            :manual-cleanup? true})))
+    (testing "with small memory-limit"
+      (test-decoder port2 url2 (make-decode-manual-handler {:raw-stream? true
+                                                            :memory-limit 12
+                                                            :manual-cleanup? true})))
+    (testing "with zero memory-limit"
+      (test-decoder port2 url2 (make-decode-manual-handler {:raw-stream? true
+                                                            :memory-limit 0
+                                                            :manual-cleanup? true})))))