WebUI: a parallel transfer engine (batches, upload sessions, ranges, six lanes)

Measured first: one libmicrohttpd thread with 64 KiB a connection held uploads to
13 MB/s, and four connections to 15.3, on a link where FTP reached 25 / 30.5 MB/s.
PS5-Suite's design was reviewed (persistent connection pool, small files per
connection, large files as pwrite ranges, big socket buffers) and an engine built
from scratch on the WebUI's HTTP (src/webui_transfer.{h,cpp}, docs/WEBUI.md):

- a pool of four server threads, 512 KiB a connection, 4 MiB socket buffers, writes
  in 1 MiB blocks; non-transfer routes under one lock as on the single thread;
- PUT /api/upload/batch: many small files in one request (a framed stream), each
  renamed into place when whole, existing names skipped and named unless replace;
- upload sessions (/api/upload/session, /part, /commit): a large file's ranges in
  any order on any number of connections, committed once covered;
- Range downloads (206); every upload still atomic, never over an existing name;
- the page plans batches, sessions and single uploads onto six lanes.

Evidence: evidence/webui-transfer-engine (base PS5 13.40, PC on Wi-Fi 6). Uploads
26.0 MB/s on one connection and 29.9 MB/s for one file in six parts (FTP: 25 /
30.5); 400 x 32 KiB at 215 files/s (FTP 105). The page against the console: 605
files in 20.1 s, all equal by FTP readback, duplicates refused, a cancelled 200 MiB
file left nothing. ThreadSanitizer under a mixed parallel load: no report.
Command: build/tmp/transfer/bench.py; node tests/webui_transfer_browser.cjs;
tools/verify.sh: all five PASS.

evidence/webui-transfer-engine/capture.json

@@ -0,0 +1,37 @@
+{
+  "step": "webui-transfer-engine",
+  "recorded": "2026-10-07T12:17:29Z",
+  "console_run": "base PS5 13.40 over Wi-Fi 6 (PC side), WebUI daemon, picker showing",
+  "revision": "working tree on 189b415 (this commit)",
+  "command": "build/tmp/transfer/bench.py <host> 6769 {baseline|engine} ...; FTP ceiling by ftplib; node tests/webui_transfer_browser.cjs against the console + FTP readback",
+  "raw_capture": "klog/webui-transfer-engine/raw.txt",
+  "raw_lines": 25,
+  "raw_bytes": 4228,
+  "lines": [
+    "link: PC on Wi-Fi 6 (rx 816 / tx 864 Mbit/s, 160 MHz), RTT ~10 ms",
+    "ftp ceiling: upload 1 x 256 MiB 25.0 MB/s; 4 parallel 30.5 MB/s; download 15.9 MB/s; 400 x 32 KiB: 46 files/s (1 conn), 105 files/s (4 conn)",
+    "before: {\"label\": \"baseline\", \"test\": \"upload 1 x 256 MiB, 1 connection\", \"seconds\": 20.59, \"files\": 1, \"MBps\": 13.0, \"files_per_s\": 0.0, \"errors\": 0, \"first_error\": null}",
+    "before: {\"label\": \"baseline\", \"test\": \"upload 4 x 256 MiB, 4 connections\", \"seconds\": 70.17, \"files\": 4, \"MBps\": 15.3, \"files_per_s\": 0.1, \"errors\": 0, \"first_error\": null}",
+    "before: {\"label\": \"baseline\", \"test\": \"upload 400 x 32 KiB, 1 keep-alive connection\", \"seconds\": 5.23, \"files\": 400, \"MBps\": 2.5, \"files_per_s\": 76.4, \"errors\": 0, \"first_error\": null}",
+    "before: {\"label\": \"baseline\", \"test\": \"upload 400 x 32 KiB, 8 keep-alive connections\", \"seconds\": 3.42, \"files\": 400, \"MBps\": 3.8, \"files_per_s\": 116.9, \"errors\": 0, \"first_error\": null}",
+    "before: {\"label\": \"baseline\", \"test\": \"download 400 x 32 KiB, 1 connection\", \"seconds\": 2.43, \"files\": 400, \"MBps\": 5.4, \"files_per_s\": 164.5, \"errors\": 0, \"first_error\": null}",
+    "before: {\"label\": \"baseline\", \"test\": \"download 400 x 32 KiB, 8 connections\", \"seconds\": 0.85, \"files\": 400, \"MBps\": 15.4, \"files_per_s\": 468.9, \"errors\": 0, \"first_error\": null}",
+    "before: {\"label\": \"baseline\", \"test\": \"download 1 x 256 MiB, 1 connection\", \"seconds\": 17.82, \"files\": 1, \"MBps\": 15.1, \"files_per_s\": 0.1, \"errors\": 0, \"first_error\": null}",
+    "before: {\"label\": \"baseline\", \"test\": \"download 4 x 256 MiB, 4 connections\", \"seconds\": 43.62, \"files\": 4, \"MBps\": 24.6, \"files_per_s\": 0.1, \"errors\": 0, \"first_error\": null}",
+    "after: {\"label\": \"engine\", \"test\": \"upload 1 x 256 MiB, 1 connection\", \"seconds\": 10.31, \"files\": 1, \"MBps\": 26.0, \"files_per_s\": 0.1, \"errors\": 0, \"first_error\": null}",
+    "after: {\"label\": \"engine\", \"test\": \"upload 4 x 256 MiB, 4 connections\", \"seconds\": 37.21, \"files\": 4, \"MBps\": 28.9, \"files_per_s\": 0.1, \"errors\": 0, \"first_error\": null}",
+    "after: {\"label\": \"engine\", \"test\": \"upload 400 x 32 KiB, 1 keep-alive connection\", \"seconds\": 4.49, \"files\": 400, \"MBps\": 2.9, \"files_per_s\": 89.0, \"errors\": 0, \"first_error\": null}",
+    "after: {\"label\": \"engine\", \"test\": \"upload 400 x 32 KiB, 8 keep-alive connections\", \"seconds\": 1.86, \"files\": 400, \"MBps\": 7.1, \"files_per_s\": 215.4, \"errors\": 0, \"first_error\": null}",
+    "after: {\"label\": \"engine\", \"test\": \"download 400 x 32 KiB, 1 connection\", \"seconds\": 1.86, \"files\": 400, \"MBps\": 7.1, \"files_per_s\": 215.5, \"errors\": 0, \"first_error\": null}",
+    "after: {\"label\": \"engine\", \"test\": \"download 400 x 32 KiB, 8 connections\", \"seconds\": 0.69, \"files\": 400, \"MBps\": 18.9, \"files_per_s\": 576.2, \"errors\": 0, \"first_error\": null}",
+    "after: {\"label\": \"engine\", \"test\": \"batch 400 x 32 KiB (100/request), 1 lane(s), fsync=1\", \"seconds\": 3.15, \"files\": 400, \"MBps\": 4.2, \"files_per_s\": 126.8, \"errors\": 0, \"first_error\": null}",
+    "after: {\"label\": \"engine\", \"test\": \"batch 400 x 32 KiB (100/request), 4 lane(s), fsync=1\", \"seconds\": 1.89, \"files\": 400, \"MBps\": 6.9, \"files_per_s\": 212.0, \"errors\": 0, \"first_error\": null}",
+    "after: {\"label\": \"engine\", \"test\": \"batch 400 x 32 KiB (100/request), 1 lane(s), fsync=0\", \"seconds\": 2.89, \"files\": 400, \"MBps\": 4.5, \"files_per_s\": 138.4, \"errors\": 0, \"first_error\": null}",
+    "after: {\"label\": \"engine\", \"test\": \"batch 400 x 32 KiB (100/request), 4 lane(s), fsync=0\", \"seconds\": 2.24, \"files\": 400, \"MBps\": 5.8, \"files_per_s\": 178.3, \"errors\": 0, \"first_error\": null}",
+    "after: {\"label\": \"engine\", \"test\": \"upload 1 x 256 MiB as 8 parts on 6 connections (commit 0.00 s)\", \"seconds\": 8.96, \"files\": 1, \"MBps\": 29.9, \"files_per_s\": 0.1, \"errors\": 0, \"first_error\": null}",
+    "after: {\"label\": \"engine\", \"test\": \"download 1 x 256 MiB, 1 connection\", \"seconds\": 13.4, \"files\": 1, \"MBps\": 20.0, \"files_per_s\": 0.1, \"errors\": 0, \"first_error\": null}",
+    "after: {\"label\": \"engine\", \"test\": \"download 4 x 256 MiB, 4 connections\", \"seconds\": 42.04, \"files\": 4, \"MBps\": 25.5, \"files_per_s\": 0.1, \"errors\": 0, \"first_error\": null}",
+    "page engine on the console: PASS: 605 files (600 small in 4 batches, 3 single, 2 large in 18 parts) in 20.1 s; duplicates refused; cancel drops the session",
+    "page engine on the console, FTP readback: 605 entries; mismatched [] ; leftovers []"
+  ]
+}

evidence/webui-transfer-engine/expectation.json

@@ -0,0 +1,15 @@
+{
+  "step": "webui-transfer-engine",
+  "note": "Parallel transfer engine (thread pool, 1 MiB blocks, batches, upload sessions, ranges). Large uploads 13 -> 26-30 MB/s (the link's FTP ceiling); small files 76/117 -> 89/215 files/s (2x FTP); the page's engine delivers 605 files whole on the console.",
+  "must_contain": [
+    "\"label\": \"baseline\", \"test\": \"upload 1 x 256 MiB, 1 connection\", \"seconds\": 20.59",
+    "\"label\": \"engine\", \"test\": \"upload 1 x 256 MiB, 1 connection\", \"seconds\": 10.31",
+    "as 8 parts on 6 connections",
+    "\"files_per_s\": 215.4",
+    "605 entries; mismatched [] ; leftovers []"
+  ],
+  "must_not_contain": [
+    "\"errors\": 1"
+  ],
+  "human_check": "none"
+}

src/webui_ps5.cpp

@@ -4,6 +4,7 @@
  */
 #include "webui_ps5.h"
 #include "webui_update.h"
+#include "webui_transfer.h"
 #include "ps5_frontend_choice.h"
 #include "../vendor/retroarch/libretro-common/include/libretro.h"
 #include <microhttpd.h>
@@ -19,7 +20,9 @@
 #include <dirent.h>
 #include <fcntl.h>
 #include <map>
+#include <memory>
 #include <mutex>
+#include <netinet/tcp.h>
 #include <new>
 #include <netinet/in.h>
 #include <string>
@@ -55,20 +58,32 @@ const Setting settings[] = {
     {"menu_show_advanced_settings", "Show advanced settings", "bool", "false", 0, 0},
     {"menu_driver", "Console menu", "menu", "xmb", 0, 0},
 };
+// The server answers on a pool of threads (ps5_webui_start). Transfers write on
+// their own; everything else (settings, folders, the updater) runs under this lock,
+// as it did on the one thread before.
+std::mutex state_lock;
+ps5_transfer::Sessions sessions;
+constexpr std::time_t session_idle_seconds = 15 * 60;
+struct BatchState;
+enum class Transfer
+{
+    none,
+    single, // PUT /api/upload
+    batch,  // PUT /api/upload/batch
+    part    // PUT /api/upload/part
+};
 struct Request
 {
-    int file = -1;
+    Transfer transfer = Transfer::none;
+    ps5_transfer::Writer writer;
+    std::unique_ptr<BatchState> batch;
+    std::shared_ptr<ps5_transfer::Session> session;
+    bool replace = false;
     std::string temporary, destination, body;
-    uint64_t received = 0, expected = 0;
+    uint64_t received = 0, expected = 0, offset = 0;
     unsigned error = 0;
     const char *message = "";
-    ~Request()
-    {
-        if (file >= 0)
-            close(file);
-        if (!temporary.empty())
-            unlink(temporary.c_str());
-    }
+    ~Request();
 };
 std::string quote(const std::string &text)
 {
@@ -242,6 +257,39 @@ bool content_path(const std::string &relative, std::string &absolute, bool new_l
     }
     return true;
 }
+// A path below content/ for a new file, its missing folders created on the way
+// (an upload of a folder's files, a batch's subfolders). The leaf is not checked.
+bool content_parents(const std::string &relative, std::string &absolute)
+{
+    if (relative.empty() || !valid_path(relative))
+        return false;
+    absolute = root_path + "/content";
+    struct stat st{};
+    if (content_stat(absolute.c_str(), &st) || !S_ISDIR(st.st_mode))
+        return false;
+    size_t start = 0;
+    for (;;)
+    {
+        const auto end = relative.find('/', start);
+        absolute += '/' + relative.substr(start, end - start);
+        if (end == std::string::npos)
+            return true;
+        if (content_stat(absolute.c_str(), &st))
+        {
+            if (errno != ENOENT || (mkdir(absolute.c_str(), 0755) && errno != EEXIST) ||
+                content_stat(absolute.c_str(), &st))
+                return false;
+        }
+        if (S_ISLNK(st.st_mode) || !S_ISDIR(st.st_mode))
+            return false;
+        start = end + 1;
+    }
+}
+bool exists(const std::string &absolute)
+{
+    struct stat st{};
+    return content_stat(absolute.c_str(), &st) == 0 || errno != ENOENT;
+}
 std::string trim(std::string s)
 {
     auto first = s.find_first_not_of(" \t\r\n");
@@ -791,17 +839,25 @@ MHD_Result list_content(MHD_Connection *c)
 // A download is read in blocks rather than handed to MHD as a descriptor: MHD sends a
 // descriptor with sendfile, which on the console stopped after its first 33 KB
 // (the daemon, 2026-10-07: curl got 33,300 of 1.9 GB and a broken transfer).
+struct DownloadSource
+{
+    int fd;
+    uint64_t start;
+};
 ssize_t read_download(void *cls, uint64_t position, char *buffer, size_t size)
 {
-    const ssize_t got = pread(static_cast<int>(reinterpret_cast<intptr_t>(cls)), buffer, size,
-                              static_cast<off_t>(position));
+    const auto *source = static_cast<DownloadSource *>(cls);
+    const ssize_t got =
+        pread(source->fd, buffer, size, static_cast<off_t>(source->start + position));
     if (got == 0)
         return MHD_CONTENT_READER_END_OF_STREAM;
     return got < 0 ? MHD_CONTENT_READER_END_WITH_ERROR : got;
 }
 void close_download(void *cls)
 {
-    close(static_cast<int>(reinterpret_cast<intptr_t>(cls)));
+    auto *source = static_cast<DownloadSource *>(cls);
+    close(source->fd);
+    delete source;
 }
 MHD_Result download(MHD_Connection *c)
 {
@@ -817,24 +873,165 @@ MHD_Result download(MHD_Connection *c)
         close(fd);
         return error(c, 400, "Choose a file to download.");
     }
-    auto *response = MHD_create_response_from_callback(
-        static_cast<uint64_t>(st.st_size), 256 * 1024, read_download,
-        reinterpret_cast<void *>(static_cast<intptr_t>(fd)), close_download);
-    if (!response)
+    // One byte range (Range: bytes=a-b), so a client can fetch a large file on several
+    // connections or resume one.
+    const uint64_t size = static_cast<uint64_t>(st.st_size);
+    uint64_t start = 0, end = size;
+    const char *range = MHD_lookup_connection_value(c, MHD_HEADER_KIND, "Range");
+    const bool partial = range && ps5_transfer::parse_range(range, size, start, end);
+    if (range && !partial && size)
     {
+        close(fd);
+        return error(c, 416, "This part of the file does not exist.");
+    }
+    auto *source = new (std::nothrow) DownloadSource{fd, start};
+    auto *response = source ? MHD_create_response_from_callback(
+                                  end - start, 1024 * 1024, read_download, source, close_download)
+                            : nullptr;
+    if (!response)
+    {
+        delete source;
         close(fd);
         return MHD_NO;
     }
+    MHD_add_response_header(response, "Accept-Ranges", "bytes");
+    if (partial)
+    {
+        const std::string content_range = "bytes " + std::to_string(start) + '-' +
+                                          std::to_string(end - 1) + '/' + std::to_string(size);
+        MHD_add_response_header(response, "Content-Range", content_range.c_str());
+    }
     const std::string disposition =
         "attachment; filename=" + quote(path.substr(path.find_last_of('/') + 1));
     MHD_add_response_header(response, "Content-Disposition", disposition.c_str());
     MHD_add_response_header(response, "Content-Type", "application/octet-stream");
     MHD_add_response_header(response, "X-Content-Type-Options", "nosniff");
     MHD_add_response_header(response, "Cache-Control", "no-store");
-    auto result = MHD_queue_response(c, 200, response);
+    auto result = MHD_queue_response(c, partial ? 206 : 200, response);
     MHD_destroy_response(response);
     return result;
 }
+// Many small files in one request (src/webui_transfer.h, the batch stream): each is
+// written under a temporary name and renamed when whole; one that exists is skipped
+// unless replace. Results are counted, not answered one by one.
+struct BatchUpload final : ps5_transfer::BatchSink
+{
+    std::string folder; // below content/, may be empty
+    bool replace = false, sync = true;
+    ps5_transfer::Writer writer;
+    std::string name, temporary, destination;
+    uint64_t size = 0;
+    unsigned written = 0, skipped = 0;
+    uint64_t bytes = 0;
+    std::vector<std::pair<std::string, std::string>> failed;
+    std::vector<std::string> skipped_names; // the first 1024
+    void skip()
+    {
+        ++skipped;
+        if (skipped_names.size() < 1024)
+            skipped_names.push_back(name);
+    }
+    bool fail(const char *why)
+    {
+        if (failed.size() < 64)
+            failed.emplace_back(name, why);
+        else
+            failed.back().second = "and more";
+        const int fd = writer.release();
+        if (fd >= 0)
+            close(fd);
+        if (!temporary.empty())
+            unlink(temporary.c_str());
+        temporary.clear();
+        return false;
+    }
+    bool begin(const std::string &record, uint64_t record_size) override
+    {
+        name = record;
+        size = record_size;
+        temporary.clear();
+        if (record_size > upload_limit ||
+            !content_parents(folder.empty() ? record : folder + '/' + record, destination))
+            return fail("invalid name");
+        if (!replace && exists(destination))
+        {
+            skip();
+            return false;
+        }
+        temporary = destination.substr(0, destination.find_last_of('/') + 1) + ".upload-" + nonce();
+        return writer.create(temporary) || fail("cannot create");
+    }
+    bool data(const char *bytes_in, size_t count) override
+    {
+        return writer.write(bytes_in, count) || fail("cannot write (storage full?)");
+    }
+    void end() override
+    {
+        const bool flushed = writer.flush();
+        const uint64_t got = writer.written();
+        const int fd = writer.release();
+        if (!flushed || fd < 0 || got != size)
+        {
+            if (fd >= 0)
+                close(fd);
+            fail("incomplete");
+            return;
+        }
+        const auto result = ps5_transfer::commit(fd, temporary, destination, replace, sync);
+        close(fd);
+        if (result == ps5_transfer::Commit::done)
+        {
+            temporary.clear();
+            ++written;
+            bytes += size;
+        }
+        else if (result == ps5_transfer::Commit::exists)
+        {
+            unlink(temporary.c_str());
+            temporary.clear();
+            skip();
+        }
+        else
+            fail("cannot finish");
+    }
+    ~BatchUpload() override
+    {
+        const int fd = writer.release();
+        if (fd >= 0)
+            close(fd);
+        if (!temporary.empty())
+            unlink(temporary.c_str());
+    }
+};
+struct BatchState
+{
+    BatchUpload upload;
+    ps5_transfer::BatchReader reader{upload};
+};
+Request::~Request()
+{
+    if (transfer == Transfer::part && session)
+    {
+        std::lock_guard<std::mutex> guard(session->lock);
+        if (--session->writers == 0 && session->closing && session->fd >= 0)
+        {
+            close(session->fd);
+            session->fd = -1;
+        }
+    }
+    if (!temporary.empty())
+        unlink(temporary.c_str());
+}
+bool content_length(MHD_Connection *c, uint64_t &length)
+{
+    const char *value = MHD_lookup_connection_value(c, MHD_HEADER_KIND, "Content-Length");
+    char *end = nullptr;
+    errno = 0;
+    if (!value || !*value || *value == '-')
+        return false;
+    length = std::strtoull(value, &end, 10);
+    return !errno && !*end;
+}
 void prepare_upload(MHD_Connection *c, Request &r)
 {
     auto fail = [&](unsigned code, const char *message)
@@ -848,8 +1045,9 @@ void prepare_upload(MHD_Connection *c, Request &r)
         fail(400, "Choose a valid content folder and filename.");
         return;
     }
-    struct stat st{};
-    if (content_stat(r.destination.c_str(), &st) == 0 || errno != ENOENT)
+    r.transfer = Transfer::single;
+    r.replace = std::strcmp(arg(c, "existing"), "replace") == 0;
+    if (!r.replace && exists(r.destination))
     {
         fail(409, "A file with this name already exists. Rename your file first.");
         return;
@@ -876,12 +1074,205 @@ void prepare_upload(MHD_Connection *c, Request &r)
     }
     r.temporary =
         r.destination.substr(0, r.destination.find_last_of('/') + 1) + ".upload-" + nonce();
-    r.file = open(r.temporary.c_str(), O_WRONLY | O_CREAT | O_EXCL | O_NOFOLLOW, 0644);
-    if (r.file < 0)
+    if (!r.writer.create(r.temporary))
     {
+        r.temporary.clear();
         fail(507, "Could not create the upload. Check console storage.");
+    }
+}
+void prepare_batch(MHD_Connection *c, Request &r)
+{
+    r.transfer = Transfer::batch;
+    if (!content_length(c, r.expected) || r.expected > upload_limit)
+    {
+        r.error = 411;
+        r.message = "A batch needs its size, at most 64 GiB.";
         return;
     }
+    auto state = std::make_unique<BatchState>();
+    state->upload.folder = arg(c, "folder");
+    state->upload.replace = std::strcmp(arg(c, "existing"), "replace") == 0;
+    state->upload.sync = std::strcmp(arg(c, "sync"), "0") != 0;
+    if (!state->upload.folder.empty() && !valid_path(state->upload.folder))
+    {
+        r.error = 400;
+        r.message = "Choose a valid content folder.";
+        return;
+    }
+    r.batch = std::move(state);
+}
+void prepare_part(MHD_Connection *c, Request &r)
+{
+    r.transfer = Transfer::part;
+    r.session = sessions.find(arg(c, "id"));
+    char *end = nullptr;
+    const char *offset = arg(c, "offset");
+    errno = 0;
+    r.offset = std::strtoull(offset, &end, 10);
+    if (!r.session)
+    {
+        r.error = 404;
+        r.message = "This upload is no longer open. Start it again.";
+        return;
+    }
+    if (!*offset || errno || *end || !content_length(c, r.expected) || r.offset > r.session->size ||
+        r.expected > r.session->size - r.offset)
+    {
+        r.session.reset();
+        r.error = 416;
+        r.message = "This part does not fit the file.";
+        return;
+    }
+    std::lock_guard<std::mutex> guard(r.session->lock);
+    if (r.session->closing || r.session->fd < 0)
+    {
+        r.session.reset();
+        r.error = 409;
+        r.message = "This upload was closed.";
+        return;
+    }
+    ++r.session->writers;
+    r.session->touched = std::time(nullptr);
+    r.writer.attach(r.session->fd, r.offset);
+}
+std::string names(const std::vector<std::string> &list)
+{
+    std::string out = "[";
+    for (const auto &name : list)
+        out += (out.size() > 1 ? "," : "") + quote(name);
+    return out + ']';
+}
+// The transfers' own routes: they run beside each other, outside state_lock.
+MHD_Result transfer_route(MHD_Connection *c, const std::string &url, const std::string &method,
+                          Request &r)
+{
+    if (r.transfer == Transfer::single)
+    {
+        if (r.received != r.expected)
+            return error(c, 400, "The upload was incomplete. Try again.");
+        const int fd = r.writer.release();
+        if (fd < 0)
+            return error(c, 507, "Could not finish writing the file. Check console storage.");
+        const auto result = ps5_transfer::commit(fd, r.temporary, r.destination, r.replace, true);
+        close(fd);
+        if (result == ps5_transfer::Commit::exists)
+            return error(c, 409, "A file with this name already exists. Rename your file first.");
+        if (result != ps5_transfer::Commit::done)
+            return error(c, 500, "Could not finish the upload.");
+        r.temporary.clear();
+        return respond(c, 201, "{\"uploaded\":true,\"bytes\":" + std::to_string(r.received) + '}');
+    }
+    if (r.transfer == Transfer::batch)
+    {
+        auto *state = r.batch.get();
+        auto &b = state->upload;
+        std::string failed = "[";
+        for (const auto &f : b.failed)
+            failed += (failed.size() > 1 ? "," : "") + std::string("{\"name\":") + quote(f.first) +
+                      ",\"error\":" + quote(f.second) + '}';
+        const bool whole = state->reader.finished() && r.received == r.expected;
+        return respond(c, whole ? 200 : 400,
+                       "{\"complete\":" + std::string(whole ? "true" : "false") + ",\"written\":" +
+                           std::to_string(b.written) + ",\"skipped\":" + std::to_string(b.skipped) +
+                           ",\"skippedNames\":" + names(b.skipped_names) + ",\"bytes\":" +
+                           std::to_string(b.bytes) + ",\"failed\":" + failed + "],\"problem\":" +
+                           quote(whole                      ? ""
+                                 : *state->reader.problem() ? state->reader.problem()
+                                                            : "The batch ended early.") +
+                           '}');
+    }
+    if (r.transfer == Transfer::part)
+    {
+        if (r.received != r.expected || !r.writer.flush())
+            return error(c, 507, "The part could not be written. Try it again.");
+        uint64_t covered;
+        {
+            // Released here, not when the request is freed after its answer: a commit
+            // sent the moment this answer arrives must find no writer left.
+            std::lock_guard<std::mutex> guard(r.session->lock);
+            r.session->add(r.offset, r.offset + r.received);
+            r.session->touched = std::time(nullptr);
+            covered = r.session->covered;
+            if (--r.session->writers == 0 && r.session->closing && r.session->fd >= 0)
+            {
+                close(r.session->fd);
+                r.session->fd = -1;
+            }
+        }
+        r.session.reset();
+        return respond(c, 200, "{\"covered\":" + std::to_string(covered) + '}');
+    }
+    if (method == "POST" && url == "/api/upload/session")
+    {
+        sessions.expire(std::time(nullptr), session_idle_seconds);
+        std::string destination;
+        char *end = nullptr;
+        const char *text = arg(c, "size");
+        errno = 0;
+        const uint64_t size = std::strtoull(text, &end, 10);
+        const bool size_ok = *text && !errno && !*end && size <= upload_limit;
+        const bool replace = std::strcmp(arg(c, "existing"), "replace") == 0;
+        if (!size_ok)
+            return error(c, 413, "Files must be 64 GiB or smaller.");
+        if (!content_parents(arg(c, "path"), destination))
+            return error(c, 400, "Choose a valid content folder and filename.");
+        if (!replace && exists(destination))
+            return error(c, 409, "A file with this name already exists. Rename your file first.");
+        struct statvfs storage{};
+        if (storage_space(storage) && size > uint64_t(storage.f_bavail) * storage.f_frsize)
+            return error(c, 507, "There is not enough free space on the console.");
+        const std::string id = nonce();
+        const char *why = "";
+        auto session = sessions.open(
+            id, destination.substr(0, destination.find_last_of('/') + 1) + ".upload-" + id,
+            destination, size, replace, why);
+        if (!session)
+            return error(c, 507, why);
+        return respond(c, 201,
+                       "{\"id\":" + quote(id) + ",\"size\":" + std::to_string(size) +
+                           ",\"partSize\":" + std::to_string(32u << 20) + '}');
+    }
+    if (method == "POST" && url == "/api/upload/commit")
+    {
+        auto session = sessions.find(arg(c, "id"));
+        if (!session)
+            return error(c, 404, "This upload is no longer open. Start it again.");
+        {
+            std::lock_guard<std::mutex> guard(session->lock);
+            if (session->writers || session->covered != session->size)
+                return respond(c, 409,
+                               "{\"error\":\"Parts are missing.\",\"covered\":" +
+                                   std::to_string(session->covered) + '}');
+        }
+        if (sessions.take(session->id) != session)
+            return error(c, 409, "This upload was already finished.");
+        const auto result = ps5_transfer::commit(session->fd, session->temporary,
+                                                 session->destination, session->replace, true);
+        close(session->fd);
+        session->fd = -1;
+        if (result != ps5_transfer::Commit::done)
+        {
+            unlink(session->temporary.c_str());
+            return result == ps5_transfer::Commit::exists
+                       ? error(c, 409,
+                               "A file with this name already exists. Rename your file first.")
+                       : error(c, 500, "Could not finish the upload.");
+        }
+        return respond(c, 201,
+                       "{\"uploaded\":true,\"bytes\":" + std::to_string(session->size) + '}');
+    }
+    if (method == "DELETE" && url == "/api/upload/session")
+    {
+        auto session = sessions.take(arg(c, "id"));
+        if (session)
+            ps5_transfer::discard(*session);
+        return respond(c, 200, "{\"cancelled\":" + std::string(session ? "true" : "false") + '}');
+    }
+    return error(c, 404, "This page was not found.");
+}
+bool is_transfer_route(const std::string &url)
+{
+    return url.rfind("/api/upload", 0) == 0;
 }
 std::string current_frontend()
 {
@@ -961,26 +1352,6 @@ MHD_Result route(MHD_Connection *c, const std::string &url, const std::string &m
                          "This folder already exists or could not be created.");
         return respond(c, 201, "{\"created\":true}");
     }
-    if (method == "PUT" && url == "/api/upload")
-    {
-        if (r.error)
-            return error(c, r.error, r.message);
-        if (r.received != r.expected)
-            return error(c, 400, "The upload was incomplete. Try again.");
-        if (fsync(r.file))
-            return error(c, 507, "Could not finish writing the file. Check console storage.");
-        close(r.file);
-        r.file = -1;
-        struct stat st{};
-        // All HTTP handlers run on one MHD thread; concurrent uploads cannot replace a completed
-        // file.
-        if (content_stat(r.destination.c_str(), &st) == 0 || errno != ENOENT)
-            return error(c, 409, "A file with this name already exists. Rename your file first.");
-        if (rename(r.temporary.c_str(), r.destination.c_str()))
-            return error(c, 500, "Could not finish the upload.");
-        r.temporary.clear();
-        return respond(c, 201, "{\"uploaded\":true,\"bytes\":" + std::to_string(r.received) + '}');
-    }
     if (method != "GET")
         return error(c, 405, "This action is not supported.");
     // Only shipped assets are reachable; no filesystem passthrough.
@@ -1019,12 +1390,14 @@ MHD_Result handle(void *, MHD_Connection *c, const char *url, const char *method
         if (write_request && (!provided || token != provided))
             return error(c, 403, "Your session expired. Reload the page and try again.");
         if (std::strcmp(method, "PUT") == 0 && std::strcmp(url, "/api/upload") == 0)
-        {
             prepare_upload(c, *r);
-            if (r->error)
-                return error(c, r->error, r->message);
-        }
-        else
+        else if (std::strcmp(method, "PUT") == 0 && std::strcmp(url, "/api/upload/batch") == 0)
+            prepare_batch(c, *r);
+        else if (std::strcmp(method, "PUT") == 0 && std::strcmp(url, "/api/upload/part") == 0)
+            prepare_part(c, *r);
+        if (r->error)
+            return error(c, r->error, r->message);
+        if (r->transfer == Transfer::none)
         {
             const char *length = MHD_lookup_connection_value(c, MHD_HEADER_KIND, "Content-Length");
             if (length && std::strtoull(length, nullptr, 10) > 16384)
@@ -1035,31 +1408,29 @@ MHD_Result handle(void *, MHD_Connection *c, const char *url, const char *method
     auto &r = *static_cast<Request *>(*context);
     if (*size)
     {
-        if (r.file >= 0 && !r.error)
+        if (r.transfer != Transfer::none && !r.error)
         {
             if (*size > r.expected - r.received)
             {
                 r.error = 413;
-                r.message = "Upload exceeded its declared file size.";
+                r.message = "Upload exceeded its declared size.";
             }
-            else
+            else if (r.transfer == Transfer::batch)
             {
-                size_t offset = 0;
-                while (offset < *size)
+                r.received += *size;
+                if (!r.batch->reader.feed(data, *size))
                 {
-                    ssize_t n = write(r.file, data + offset, *size - offset);
-                    if (n < 0 && errno == EINTR)
-                        continue;
-                    if (n <= 0)
-                    {
-                        r.error = 507;
-                        r.message = "The console could not write the upload. Check free space.";
-                        break;
-                    }
-                    offset += size_t(n);
-                    r.received += uint64_t(n);
+                    r.error = 400;
+                    r.message = r.batch->reader.problem();
                 }
             }
+            else if (!r.writer.write(data, *size))
+            {
+                r.error = 507;
+                r.message = "The console could not write the upload. Check free space.";
+            }
+            else
+                r.received += *size;
         }
         else if (!r.error)
         {
@@ -1076,10 +1447,26 @@ MHD_Result handle(void *, MHD_Connection *c, const char *url, const char *method
         // never during a body callback. Drain failed streams without buffering.
         return MHD_YES;
     }
-    if (r.error)
+    if (r.error && !(r.transfer == Transfer::batch && r.batch && r.error == 400))
         return error(c, r.error, r.message);
+    if (r.transfer != Transfer::none || is_transfer_route(url))
+        return transfer_route(c, url, method, r);
+    std::lock_guard<std::mutex> guard(state_lock);
     return route(c, url, method, r);
 }
+constexpr unsigned transfer_threads = 4;
+// Large socket buffers for each connection: a lane keeps the link full on its own.
+void tune_connection(void *, MHD_Connection *c, void **, MHD_ConnectionNotificationCode code)
+{
+    if (code != MHD_CONNECTION_NOTIFY_STARTED)
+        return;
+    const auto *info = MHD_get_connection_info(c, MHD_CONNECTION_INFO_CONNECTION_FD);
+    if (!info)
+        return;
+    int buffer = 4 << 20;
+    setsockopt(info->connect_fd, SOL_SOCKET, SO_RCVBUF, &buffer, sizeof buffer);
+    setsockopt(info->connect_fd, SOL_SOCKET, SO_SNDBUF, &buffer, sizeof buffer);
+}
 void completed(void *, MHD_Connection *, void **context, MHD_RequestTerminationCode)
 {
     delete static_cast<Request *>(*context);
@@ -1233,12 +1620,16 @@ bool ps5_webui_start(const char *root, unsigned short port, bool apply_settings)
     listen_port = port;
     token = nonce();
     mkdir((root_path + "/content").c_str(), 0755);
+    // A pool of threads, so parallel connections progress side by side (one thread
+    // held a 4-connection upload to 15 MB/s, 2026-10-07); 512 KiB a connection, so a
+    // body arrives in large pieces; up to 32 connections for the page's lanes.
     web_daemon = MHD_start_daemon(
         MHD_USE_INTERNAL_POLLING_THREAD | MHD_USE_ITC | MHD_USE_ERROR_LOG, port, nullptr, nullptr,
-        handle, nullptr, MHD_OPTION_CONNECTION_LIMIT, unsigned(8), MHD_OPTION_CONNECTION_TIMEOUT,
-        unsigned(30), MHD_OPTION_CONNECTION_MEMORY_LIMIT, size_t(65536),
-        MHD_OPTION_THREAD_STACK_SIZE, size_t(256 * 1024), MHD_OPTION_NOTIFY_COMPLETED, completed,
-        nullptr, MHD_OPTION_END);
+        handle, nullptr, MHD_OPTION_THREAD_POOL_SIZE, unsigned(transfer_threads),
+        MHD_OPTION_CONNECTION_LIMIT, unsigned(32), MHD_OPTION_CONNECTION_TIMEOUT, unsigned(30),
+        MHD_OPTION_CONNECTION_MEMORY_LIMIT, size_t(512 * 1024), MHD_OPTION_THREAD_STACK_SIZE,
+        size_t(256 * 1024), MHD_OPTION_NOTIFY_COMPLETED, completed, nullptr,
+        MHD_OPTION_NOTIFY_CONNECTION, tune_connection, nullptr, MHD_OPTION_END);
     std::fprintf(stderr, "webui: %s port=%u\n", web_daemon ? "listening" : "unavailable", port);
     return web_daemon != nullptr;
 }
@@ -1250,5 +1641,6 @@ void ps5_webui_stop()
         web_daemon = nullptr;
         std::fprintf(stderr, "webui: stopped\n");
     }
+    sessions.clear();
     ps5_update::stop();
 }

src/webui_transfer.cpp

@@ -0,0 +1,373 @@
+/* Copyright (C) 2026 Mihawk; SPDX-License-Identifier: GPL-3.0-or-later */
+/* The WebUI's transfer engine (src/webui_transfer.h says what each part is for). */
+#include "webui_transfer.h"
+
+#include <algorithm>
+#include <cerrno>
+#include <cstdlib>
+#include <cstring>
+#include <iterator>
+#include <fcntl.h>
+#include <sys/stat.h>
+#include <unistd.h>
+
+namespace ps5_transfer
+{
+Writer::~Writer()
+{
+    if (owned_ && fd_ >= 0)
+        close(fd_);
+}
+
+bool Writer::create(const std::string &path)
+{
+    fd_ = open(path.c_str(), O_WRONLY | O_CREAT | O_EXCL | O_NOFOLLOW, 0644);
+    owned_ = fd_ >= 0;
+    offset_ = written_ = 0;
+    return owned_;
+}
+
+void Writer::attach(int fd, uint64_t offset)
+{
+    fd_ = fd;
+    owned_ = false;
+    offset_ = offset;
+    written_ = 0;
+}
+
+bool Writer::write(const char *data, size_t size)
+{
+    if (fd_ < 0)
+        return false;
+    if (buffer_.empty())
+        buffer_.resize(write_block);
+    while (size)
+    {
+        /* A piece as large as the block goes straight to the file. */
+        if (used_ == 0 && size >= write_block)
+        {
+            const size_t whole = size - size % write_block;
+            const ssize_t n = pwrite(fd_, data, whole, static_cast<off_t>(offset_ + written_));
+            if (n < 0 && errno == EINTR)
+                continue;
+            if (n <= 0)
+                return false;
+            written_ += uint64_t(n);
+            data += n;
+            size -= size_t(n);
+            continue;
+        }
+        const size_t take = std::min(size, write_block - used_);
+        std::memcpy(buffer_.data() + used_, data, take);
+        used_ += take;
+        data += take;
+        size -= take;
+        if (used_ == write_block && !flush())
+            return false;
+    }
+    return true;
+}
+
+bool Writer::flush()
+{
+    size_t done = 0;
+    while (done < used_)
+    {
+        const ssize_t n = pwrite(fd_, buffer_.data() + done, used_ - done,
+                                 static_cast<off_t>(offset_ + written_));
+        if (n < 0 && errno == EINTR)
+            continue;
+        if (n <= 0)
+            return false;
+        done += size_t(n);
+        written_ += uint64_t(n);
+    }
+    used_ = 0;
+    return true;
+}
+
+int Writer::release()
+{
+    const int fd = flush() ? fd_ : -1;
+    if (fd < 0 && owned_ && fd_ >= 0)
+        close(fd_);
+    fd_ = -1;
+    owned_ = false;
+    return fd;
+}
+
+namespace
+{
+std::mutex commit_lock;
+}
+
+Commit commit(int data_fd, const std::string &temporary, const std::string &destination,
+              bool replace, bool sync)
+{
+    if (data_fd >= 0 && sync && fsync(data_fd))
+        return Commit::failed;
+    std::lock_guard<std::mutex> guard(commit_lock);
+    if (!replace)
+    {
+        /* Opened, not stat'ed: a title cannot use lstat reliably (src/webui_ps5.cpp). */
+        const int existing = open(destination.c_str(), O_RDONLY | O_NOFOLLOW | O_NONBLOCK);
+        if (existing >= 0 || errno != ENOENT)
+        {
+            if (existing >= 0)
+                close(existing);
+            return Commit::exists;
+        }
+    }
+    return rename(temporary.c_str(), destination.c_str()) == 0 ? Commit::done : Commit::failed;
+}
+
+bool BatchReader::fail(const char *why)
+{
+    if (accepted_)
+        sink_.end();
+    accepted_ = false;
+    state_ = State::bad;
+    problem_ = why;
+    return false;
+}
+
+bool BatchReader::feed(const char *data, size_t size)
+{
+    while (size || (state_ == State::data && remaining_ == 0))
+    {
+        switch (state_)
+        {
+        case State::bad:
+            return false;
+        case State::done:
+            return size == 0 || fail("The batch has data after its end.");
+        case State::name_length:
+        case State::size:
+        {
+            const size_t want = state_ == State::name_length ? 2 : 8;
+            const size_t take = std::min(size, want - have_);
+            std::memcpy(header_ + have_, data, take);
+            have_ += take;
+            data += take;
+            size -= take;
+            if (have_ < want)
+                break;
+            have_ = 0;
+            if (state_ == State::name_length)
+            {
+                name_length_ = size_t(header_[0]) | size_t(header_[1]) << 8;
+                if (name_length_ == 0)
+                    state_ = State::done;
+                else if (name_length_ > 1024)
+                    return fail("A file name in the batch is too long.");
+                else
+                {
+                    name_.clear();
+                    state_ = State::name;
+                }
+            }
+            else
+            {
+                remaining_ = 0;
+                for (int i = 7; i >= 0; --i)
+                    remaining_ = remaining_ << 8 | header_[i];
+                accepted_ = sink_.begin(name_, remaining_);
+                ++records_;
+                state_ = State::data;
+            }
+            break;
+        }
+        case State::name:
+        {
+            const size_t take = std::min(size, name_length_ - name_.size());
+            name_.append(data, take);
+            data += take;
+            size -= take;
+            if (name_.size() == name_length_)
+                state_ = State::size;
+            break;
+        }
+        case State::data:
+        {
+            const size_t take = size_t(std::min<uint64_t>(size, remaining_));
+            if (accepted_ && take && !sink_.data(data, take))
+                accepted_ = false; /* the sink keeps why; the rest is drained */
+            data += take;
+            size -= take;
+            remaining_ -= take;
+            if (remaining_ == 0)
+            {
+                if (accepted_)
+                    sink_.end();
+                accepted_ = false;
+                state_ = State::name_length;
+            }
+            break;
+        }
+        }
+    }
+    return state_ != State::bad;
+}
+
+void Session::add(uint64_t start, uint64_t end)
+{
+    if (start >= end)
+        return;
+    auto next = ranges.upper_bound(start);
+    if (next != ranges.begin())
+    {
+        auto previous = std::prev(next);
+        if (previous->second >= start)
+        {
+            start = previous->first;
+            end = std::max(end, previous->second);
+            covered -= previous->second - previous->first;
+            ranges.erase(previous);
+        }
+    }
+    next = ranges.lower_bound(start);
+    while (next != ranges.end() && next->first <= end)
+    {
+        end = std::max(end, next->second);
+        covered -= next->second - next->first;
+        next = ranges.erase(next);
+    }
+    ranges[start] = end;
+    covered += end - start;
+}
+
+std::shared_ptr<Session> Sessions::open(const std::string &id, const std::string &temporary,
+                                        const std::string &destination, uint64_t size, bool replace,
+                                        const char *&why)
+{
+    std::lock_guard<std::mutex> guard(lock_);
+    if (sessions_.size() >= limit)
+    {
+        why = "Too many large uploads at once. Wait for one to finish.";
+        return nullptr;
+    }
+    auto session = std::make_shared<Session>();
+    session->id = id;
+    session->temporary = temporary;
+    session->destination = destination;
+    session->size = size;
+    session->replace = replace;
+    session->touched = std::time(nullptr);
+    session->fd = ::open(temporary.c_str(), O_RDWR | O_CREAT | O_EXCL | O_NOFOLLOW, 0644);
+    if (session->fd < 0)
+    {
+        why = "Could not create the upload. Check console storage.";
+        return nullptr;
+    }
+    /* Sized now, so ranges land anywhere; sparse until written. */
+    if (ftruncate(session->fd, static_cast<off_t>(size)))
+    {
+        close(session->fd);
+        unlink(temporary.c_str());
+        why = "There is not enough free space on the console.";
+        return nullptr;
+    }
+    sessions_[id] = session;
+    return session;
+}
+
+std::shared_ptr<Session> Sessions::find(const std::string &id)
+{
+    std::lock_guard<std::mutex> guard(lock_);
+    auto it = sessions_.find(id);
+    return it == sessions_.end() ? nullptr : it->second;
+}
+
+std::shared_ptr<Session> Sessions::take(const std::string &id)
+{
+    std::lock_guard<std::mutex> guard(lock_);
+    auto it = sessions_.find(id);
+    if (it == sessions_.end())
+        return nullptr;
+    auto session = it->second;
+    sessions_.erase(it);
+    return session;
+}
+
+void discard(Session &session)
+{
+    std::lock_guard<std::mutex> guard(session.lock);
+    session.closing = true;
+    if (session.writers == 0 && session.fd >= 0)
+    {
+        close(session.fd);
+        session.fd = -1;
+    }
+    unlink(session.temporary.c_str());
+}
+
+void Sessions::expire(std::time_t now, std::time_t seconds)
+{
+    std::vector<std::shared_ptr<Session>> stale;
+    {
+        std::lock_guard<std::mutex> guard(lock_);
+        for (auto it = sessions_.begin(); it != sessions_.end();)
+        {
+            std::lock_guard<std::mutex> session_guard(it->second->lock);
+            if (it->second->writers == 0 && now - it->second->touched > seconds)
+            {
+                stale.push_back(it->second);
+                it = sessions_.erase(it);
+            }
+            else
+                ++it;
+        }
+    }
+    for (auto &session : stale)
+        discard(*session);
+}
+
+void Sessions::clear()
+{
+    std::map<std::string, std::shared_ptr<Session>> all;
+    {
+        std::lock_guard<std::mutex> guard(lock_);
+        all.swap(sessions_);
+    }
+    for (auto &entry : all)
+        discard(*entry.second);
+}
+
+size_t Sessions::count()
+{
+    std::lock_guard<std::mutex> guard(lock_);
+    return sessions_.size();
+}
+
+bool parse_range(const char *header, uint64_t size, uint64_t &start, uint64_t &end)
+{
+    if (!header || std::strncmp(header, "bytes=", 6) != 0 || size == 0)
+        return false;
+    const char *p = header + 6;
+    char *after = nullptr;
+    if (*p == '-')
+    {
+        /* The last n bytes. */
+        const uint64_t last = std::strtoull(p + 1, &after, 10);
+        if (after == p + 1 || *after || last == 0)
+            return false;
+        start = last >= size ? 0 : size - last;
+        end = size;
+        return true;
+    }
+    start = std::strtoull(p, &after, 10);
+    if (after == p || *after != '-' || start >= size)
+        return false;
+    p = after + 1;
+    if (*p == '\0')
+    {
+        end = size;
+        return true;
+    }
+    const uint64_t last = std::strtoull(p, &after, 10);
+    if (*after || last < start)
+        return false;
+    end = std::min(last + 1, size);
+    return true;
+}
+} // namespace ps5_transfer

src/webui_transfer.h

@@ -0,0 +1,177 @@
+/* Copyright (C) 2026 Mihawk; SPDX-License-Identifier: GPL-3.0-or-later */
+/* The WebUI's transfer engine, apart from HTTP (src/webui_ps5.cpp wires it).
+ *
+ * Three ways in, all ending the same way: a file is written under a hidden
+ * temporary name beside its destination and renamed into place only when every
+ * byte is there, so a transfer cut short never leaves a partial file behind.
+ *
+ *   - one file, one request (PUT /api/upload), written in 1 MiB blocks;
+ *   - many small files, one request (PUT /api/upload/batch): a stream of records,
+ *     so thousands of files cost one round trip per batch instead of one per file;
+ *   - one large file, several requests at once (an upload session): byte ranges
+ *     arrive in any order on as many connections as the client opens, each written
+ *     at its own offset, and the file is committed once its ranges cover it.
+ *
+ * The server answers on several threads (a pool), so these are thread-safe: a
+ * session's ranges under its own lock, renames under one commit lock. */
+#pragma once
+#include <cstddef>
+#include <cstdint>
+#include <ctime>
+#include <map>
+#include <memory>
+#include <mutex>
+#include <string>
+#include <vector>
+
+namespace ps5_transfer
+{
+constexpr size_t write_block = 1u << 20;
+
+/* Writes a stream through a block buffer, from a start offset (pwrite), so one file
+ * can take several writers at different offsets. */
+class Writer
+{
+  public:
+    Writer() = default;
+    Writer(const Writer &) = delete;
+    Writer &operator=(const Writer &) = delete;
+    ~Writer();
+    /* A new file, created exclusive; the writer owns and closes it. */
+    bool create(const std::string &path);
+    /* Someone else's descriptor (a session's), from offset. */
+    void attach(int fd, uint64_t offset);
+    bool write(const char *data, size_t size);
+    bool flush();
+    int fd() const
+    {
+        return fd_;
+    }
+    uint64_t written() const
+    {
+        return written_;
+    }
+    /* Flushes, then gives the descriptor up (the caller closes an owned one). */
+    int release();
+
+  private:
+    int fd_ = -1;
+    bool owned_ = false;
+    uint64_t offset_ = 0, written_ = 0;
+    std::vector<char> buffer_;
+    size_t used_ = 0;
+};
+
+/* Moves a finished temporary file to its destination: never over an existing file
+ * unless replace. Every rename goes through one lock, so two writers of the same
+ * name cannot both pass the existence check. data_fd (or -1) is synced first. */
+enum class Commit
+{
+    done,
+    exists,
+    failed
+};
+Commit commit(int data_fd, const std::string &temporary, const std::string &destination,
+              bool replace, bool sync);
+
+/* The batch stream: records of
+ *     u16 name length (little-endian), name (a path below the batch folder),
+ *     u64 size, size bytes of data,
+ * ended by a name length of 0. The reader takes the body in whatever pieces it
+ * arrives in and calls the sink at each record's start, data and end. */
+class BatchSink
+{
+  public:
+    virtual ~BatchSink() = default;
+    /* A record starts: false skips its data (the reason is the sink's to keep). */
+    virtual bool begin(const std::string &name, uint64_t size) = 0;
+    virtual bool data(const char *bytes, size_t size) = 0;
+    virtual void end() = 0;
+};
+class BatchReader
+{
+  public:
+    explicit BatchReader(BatchSink &sink) : sink_(sink)
+    {
+    }
+    /* false once the stream is malformed (an empty or oversized name, data past the
+     * end marker); the files before it stand. */
+    bool feed(const char *data, size_t size);
+    bool finished() const
+    {
+        return state_ == State::done;
+    }
+    const char *problem() const
+    {
+        return problem_;
+    }
+    unsigned records() const
+    {
+        return records_;
+    }
+
+  private:
+    enum class State
+    {
+        name_length,
+        name,
+        size,
+        data,
+        done,
+        bad
+    };
+    BatchSink &sink_;
+    State state_ = State::name_length;
+    unsigned char header_[8] = {};
+    size_t have_ = 0, name_length_ = 0;
+    std::string name_;
+    uint64_t remaining_ = 0;
+    bool accepted_ = false;
+    unsigned records_ = 0;
+    const char *problem_ = "";
+    bool fail(const char *why);
+};
+
+/* One large file arriving in ranges (an upload session). */
+struct Session
+{
+    std::string id, temporary, destination;
+    uint64_t size = 0;
+    bool replace = false;
+    int fd = -1;
+    std::mutex lock;
+    std::map<uint64_t, uint64_t> ranges; /* start -> end, merged */
+    uint64_t covered = 0;
+    unsigned writers = 0;
+    bool closing = false;
+    std::time_t touched = 0;
+    /* Records [start, end) as received; under lock. */
+    void add(uint64_t start, uint64_t end);
+};
+class Sessions
+{
+  public:
+    /* A new session, its temporary file created and sized; nullptr with why set. */
+    std::shared_ptr<Session> open(const std::string &id, const std::string &temporary,
+                                  const std::string &destination, uint64_t size, bool replace,
+                                  const char *&why);
+    std::shared_ptr<Session> find(const std::string &id);
+    /* Takes the session out (commit or abort); its file is the caller's then. */
+    std::shared_ptr<Session> take(const std::string &id);
+    /* Drops sessions idle for longer than seconds (their temporary files deleted). */
+    void expire(std::time_t now, std::time_t seconds);
+    /* Every session gone, as the server stops. */
+    void clear();
+    size_t count();
+    static constexpr size_t limit = 16;
+
+  private:
+    std::mutex lock_;
+    std::map<std::string, std::shared_ptr<Session>> sessions_;
+};
+/* Closes and deletes a session's file (abort, expiry). */
+void discard(Session &session);
+
+/* "bytes=a-b" (one range) against a file's size: false when absent or unusable. */
+bool parse_range(const char *header, uint64_t size, uint64_t &start, uint64_t &end);
+} // namespace ps5_transfer

tests/test_webui.py

@@ -28,7 +28,7 @@ class WebUI(unittest.TestCase):
         subprocess.run(['c++', '-std=c++17', '-O1', '-g', '-pthread',
                         '-I'+str(ROOT / '.deps/webui/libmicrohttpd-1.0.10/src/include'),
                         '-I'+str(ROOT / '.deps/native/zlib/zlib-1.3.2/contrib/minizip'), '-I'+str(ROOT / 'vendor/retroarch/deps/mbedtls'),
-                        'tests/webui_server_main.cpp', 'src/webui_ps5.cpp', 'src/webui_update.cpp', archive, update, '-lz', '-o', str(cls.binary)], cwd=ROOT, check=True)
+                        'tests/webui_server_main.cpp', 'src/webui_ps5.cpp', 'src/webui_transfer.cpp', 'src/webui_update.cpp', archive, update, '-lz', '-o', str(cls.binary)], cwd=ROOT, check=True)
         with socket.socket() as s:
             s.bind(('127.0.0.1', 0)); cls.port = s.getsockname()[1]
         cls.process = subprocess.Popen([str(cls.binary), str(cls.root), str(cls.port)], stdout=subprocess.DEVNULL, stderr=subprocess.DEVNULL)
@@ -328,3 +328,117 @@ class WebUI(unittest.TestCase):
             time.sleep(.02)
         self.assertFalse(list((self.root / 'content').glob('.upload-*')))
         self.assertFalse((self.root / 'content/unfinished.iso').exists())
+
+    # The transfer engine (src/webui_transfer.h): batches, sessions, ranges, parallel.
+    @staticmethod
+    def batch(files, end=True):
+        out = b''
+        for name, data in files:
+            raw = name.encode()
+            out += len(raw).to_bytes(2, 'little') + raw + len(data).to_bytes(8, 'little') + data
+        return out + (b'\0\0' if end else b'')
+
+    def test_batch_upload_many_small_files(self):
+        files = [(f'media/{system}/{i:03}.png', bytes([i % 251]) * (i * 37 % 5000))
+                 for system in ('snes', 'nes') for i in range(150)]
+        status, _, body = self.request('PUT', '/api/upload/batch?folder=batch-a', self.batch(files))
+        result = json.loads(body)
+        self.assertEqual((status, result['written'], result['skipped'], result['failed']), (200, 300, 0, []))
+        for name, data in files:
+            self.assertEqual((self.root / 'content/batch-a' / name).read_bytes(), data)
+        # Existing files are skipped unless replace; a replace writes them again.
+        changed = [(files[0][0], b'new'), ('media/snes/extra.png', b'x')]
+        result = json.loads(self.request('PUT', '/api/upload/batch?folder=batch-a', self.batch(changed))[2])
+        self.assertEqual((result['written'], result['skipped']), (1, 1))
+        self.assertEqual((self.root / 'content/batch-a' / files[0][0]).read_bytes(), files[0][1])
+        result = json.loads(self.request('PUT', '/api/upload/batch?folder=batch-a&existing=replace&sync=0',
+                                         self.batch(changed))[2])
+        self.assertEqual((result['written'], result['skipped']), (2, 0))
+        self.assertEqual((self.root / 'content/batch-a' / files[0][0]).read_bytes(), b'new')
+        self.assertFalse(list((self.root / 'content/batch-a').rglob('.upload-*')))
+
+    def test_batch_refuses_bad_names_and_bad_streams(self):
+        bad = [('../escape.png', b'a'), ('.hidden', b'b'), ('a//b', b'c'), ('ok/fine.png', b'd'), ('/abs', b'e')]
+        status, _, body = self.request('PUT', '/api/upload/batch?folder=batch-b', self.batch(bad))
+        result = json.loads(body)
+        self.assertEqual((status, result['written'], len(result['failed'])), (200, 1, 4))
+        self.assertFalse((self.root / 'escape.png').exists())
+        self.assertEqual((self.root / 'content/batch-b/ok/fine.png').read_bytes(), b'd')
+        # Cut short: the files before the cut stand, the one cut is not left behind.
+        stream = self.batch([('whole.png', b'1' * 100)], end=False)
+        stream += (8).to_bytes(2, 'little') + b'half.png' + (1000).to_bytes(8, 'little') + b'2' * 10
+        status, _, body = self.request('PUT', '/api/upload/batch?folder=batch-c', stream)
+        result = json.loads(body)
+        self.assertEqual((status, result['complete'], result['written']), (400, False, 1))
+        self.assertFalse((self.root / 'content/batch-c/half.png').exists())
+        self.assertFalse(list((self.root / 'content/batch-c').glob('.upload-*')))
+        # Data past the end marker, and a folder outside content, are refused.
+        self.assertEqual(self.request('PUT', '/api/upload/batch?folder=batch-d', self.batch([]) + b'junk')[0], 400)
+        self.assertEqual(self.request('PUT', '/api/upload/batch?folder=../x', self.batch([]))[0], 400)
+        self.assertEqual(self.request('PUT', '/api/upload/batch?folder=x', self.batch([]),
+                                      headers={'X-RetroArch-Token': 'wrong'})[0], 403)
+
+    def test_session_parts_in_parallel_any_order(self):
+        import os, random, threading
+        data = os.urandom(5 * 1024 * 1024 + 123)
+        status, _, body = self.request('POST', '/api/upload/session?path=' + quote('big/game.iso') + f'&size={len(data)}')
+        self.assertEqual(status, 201)
+        sid = json.loads(body)['id']
+        # Missing parts: the commit says what is covered and leaves the session open.
+        status, _, body = self.request('POST', f'/api/upload/commit?id={sid}')
+        self.assertEqual((status, json.loads(body)['covered']), (409, 0))
+        part = 700 * 1024
+        ranges = [(o, min(o + part + 4096, len(data))) for o in range(0, len(data), part)]  # overlapping
+        random.Random(7).shuffle(ranges)
+        results = []
+        def send(chunk):
+            for start, end in chunk:
+                results.append(self.request('PUT', f'/api/upload/part?id={sid}&offset={start}', data[start:end])[0])
+        threads = [threading.Thread(target=send, args=(ranges[i::6],)) for i in range(6)]
+        [t.start() for t in threads]; [t.join() for t in threads]
+        self.assertEqual(set(results), {200})
+        self.assertFalse((self.root / 'content/big/game.iso').exists())
+        status, _, body = self.request('POST', f'/api/upload/commit?id={sid}')
+        self.assertEqual((status, json.loads(body)['bytes']), (201, len(data)))
+        self.assertEqual((self.root / 'content/big/game.iso').read_bytes(), data)
+        self.assertFalse(list((self.root / 'content/big').glob('.upload-*')))
+        # The session is gone; a second commit, or a part, finds nothing.
+        self.assertEqual(self.request('POST', f'/api/upload/commit?id={sid}')[0], 404)
+        self.assertEqual(self.request('PUT', f'/api/upload/part?id={sid}&offset=0', b'x')[0], 404)
+        # The name now exists: a new session for it is refused unless replace.
+        self.assertEqual(self.request('POST', '/api/upload/session?path=big/game.iso&size=5')[0], 409)
+
+    def test_session_bounds_abort_and_range_download(self):
+        status, _, body = self.request('POST', '/api/upload/session?path=abort.bin&size=10')
+        sid = json.loads(body)['id']
+        self.assertEqual(self.request('PUT', f'/api/upload/part?id={sid}&offset=8', b'abc')[0], 416)
+        self.assertEqual(self.request('PUT', f'/api/upload/part?id={sid}&offset=x', b'a')[0], 416)
+        self.assertEqual(self.request('PUT', f'/api/upload/part?id={sid}&offset=0', b'0123456789')[0], 200)
+        self.assertEqual(json.loads(self.request('DELETE', f'/api/upload/session?id={sid}')[2])['cancelled'], True)
+        self.assertFalse((self.root / 'content/abort.bin').exists())
+        self.assertFalse(list((self.root / 'content').glob('.upload-*')))
+        self.assertEqual(self.request('POST', '/api/upload/session?path=../x&size=1')[0], 400)
+        # Ranges of a download: a slice, an open end, the last bytes, and one past the end.
+        (self.root / 'content/range.bin').write_bytes(bytes(range(256)) * 4)
+        status, headers, body = self.request('GET', '/api/download?path=range.bin', headers={'Range': 'bytes=10-19'})
+        self.assertEqual((status, body, headers['Content-Range']), (206, bytes(range(10, 20)), 'bytes 10-19/1024'))
+        self.assertEqual(self.request('GET', '/api/download?path=range.bin', headers={'Range': 'bytes=1000-'})[2],
+                         (bytes(range(256)) * 4)[1000:])
+        self.assertEqual(self.request('GET', '/api/download?path=range.bin', headers={'Range': 'bytes=-4'})[2],
+                         bytes(range(252, 256)))
+        self.assertEqual(self.request('GET', '/api/download?path=range.bin', headers={'Range': 'bytes=2000-'})[0], 416)
+        status, headers, body = self.request('GET', '/api/download?path=range.bin')
+        self.assertEqual((status, len(body), headers['Accept-Ranges']), (200, 1024, 'bytes'))
+
+    def test_parallel_uploads_of_one_name_keep_one(self):
+        import threading
+        results = []
+        def send(i):
+            results.append(self.request('PUT', '/api/upload?path=race.bin', str(i).encode() * 100000)[0])
+        threads = [threading.Thread(target=send, args=(i,)) for i in range(8)]
+        [t.start() for t in threads]; [t.join() for t in threads]
+        self.assertEqual(results.count(201), 1)
+        self.assertEqual(results.count(409), 7)
+        content = (self.root / 'content/race.bin').read_bytes()
+        self.assertEqual(len(set(content)), 1)
+        self.assertFalse(list((self.root / 'content').glob('.upload-*')))

tests/test_webui_daemon.py

@@ -47,7 +47,7 @@ class WebuiDaemon(unittest.TestCase):
                         '-I' + str(ROOT / 'vendor/retroarch/deps/mbedtls'),
                         f'-DPS5_WEBUI_HOMEBREW="{cls.homebrew}"', f'-DPS5_WEBUI_LINK_PORT={cls.link_port}',
                         f'-DPS5_WEBUI_HTTP_PORT={cls.http_port}', f'-DPS5_WEBUI_IDLE_SECONDS={IDLE}',
-                        'daemon/webui_daemon.cpp', 'src/webui_ps5.cpp', 'src/webui_update.cpp', http, update,
+                        'daemon/webui_daemon.cpp', 'src/webui_ps5.cpp', 'src/webui_transfer.cpp', 'src/webui_update.cpp', http, update,
                         '-lz', '-o', str(cls.binary)], cwd=ROOT, check=True)
 
     @classmethod

tests/test_webui_transfer_browser.py

@@ -0,0 +1,65 @@
+"""The page's transfer engine in Chromium (tests/webui_transfer_browser.cjs) against
+the native server, then the files on disk compared with what the page made."""
+from pathlib import Path
+import os
+import shutil
+import socket
+import subprocess
+import tempfile
+import time
+import unittest
+
+ROOT = Path(__file__).resolve().parent.parent
+PLAYWRIGHT = os.environ.get('PLAYWRIGHT_PATH') or next(
+    (str(p) for p in (Path.home() / '.cache/codex-runtimes/codex-primary-runtime/dependencies/node/node_modules/playwright-core',
+                      Path('/usr/lib/chatgpt/resources/cua_node/lib/node_modules/playwright-core')) if p.exists()), '')
+
+
+def pattern(size, seed):
+    data = bytearray(size)
+    for i in range(0, size, 4096):
+        data[i] = (seed + i) & 255
+    return bytes(data)
+
+
+@unittest.skipUnless(PLAYWRIGHT and shutil.which('node') and Path('/usr/bin/chromium').exists(), 'needs node, Playwright and Chromium')
+class TransferBrowser(unittest.TestCase):
+    def test_page_engine(self):
+        http = subprocess.check_output(['bash', 'tools/build-webui-http.sh', 'host'], cwd=ROOT, text=True).strip()
+        update = subprocess.check_output(['bash', 'tools/build-webui-update.sh', 'host'], cwd=ROOT, text=True).strip()
+        with tempfile.TemporaryDirectory() as temp:
+            root = Path(temp) / 'title'
+            for name in ('config', 'content', 'webui'):
+                (root / name).mkdir(parents=True)
+            shutil.copytree(ROOT / 'webui', root / 'webui', dirs_exist_ok=True)
+            binary = Path(temp) / 'server'
+            subprocess.run(['c++', '-std=c++17', '-O1', '-pthread',
+                            '-I' + str(ROOT / '.deps/webui/libmicrohttpd-1.0.10/src/include'),
+                            '-I' + str(ROOT / '.deps/native/zlib/zlib-1.3.2/contrib/minizip'),
+                            '-I' + str(ROOT / 'vendor/retroarch/deps/mbedtls'),
+                            'tests/webui_server_main.cpp', 'src/webui_ps5.cpp', 'src/webui_transfer.cpp',
+                            'src/webui_update.cpp', http, update, '-lz', '-o', str(binary)], cwd=ROOT, check=True)
+            with socket.socket() as s:
+                s.bind(('127.0.0.1', 0)); port = s.getsockname()[1]
+            server = subprocess.Popen([str(binary), str(root), str(port)], stdout=subprocess.DEVNULL, stderr=subprocess.DEVNULL)
+            try:
+                time.sleep(0.5)
+                run = subprocess.run(['node', 'tests/webui_transfer_browser.cjs'], cwd=ROOT, capture_output=True, text=True, timeout=600,
+                                     env={**os.environ, 'PLAYWRIGHT_PATH': PLAYWRIGHT, 'WEBUI_TEST_URL': f'http://127.0.0.1:{port}'})
+                self.assertEqual(run.returncode, 0, run.stdout + run.stderr)
+                print(run.stdout.strip())
+            finally:
+                server.terminate(); server.wait(timeout=5)
+            folder = root / 'content/engine'
+            for i in range(600):
+                self.assertEqual((folder / f'shot-{i:04}.png').read_bytes(), pattern(1000 + (i * 997) % 60000, i))
+            for i in range(3):
+                self.assertEqual((folder / f'medium-{i}.bin').read_bytes(), pattern(9 * 1024 * 1024 + i, 7 + i))
+            for i in range(2):
+                self.assertEqual((folder / f'large-{i}.iso').read_bytes(), pattern(150 * 1024 * 1024 + 4097 * i, 31 + i))
+            self.assertFalse((folder / 'cancel.iso').exists())
+            self.assertEqual(list(folder.glob('.upload-*')), [])
+
+
+if __name__ == '__main__':
+    unittest.main()

tests/webui_transfer_browser.cjs

@@ -0,0 +1,59 @@
+/* The page's transfer engine in Chromium against a native webui_server_main fixture
+ * (tests/test_webui_transfer_browser.py starts it): many small files go as batches,
+ * large ones as parallel parts, the rest one to a request; all arrive whole, names
+ * that exist are refused, and a cancelled large file leaves nothing behind. */
+const assert = require('node:assert/strict');
+const { chromium } = require(process.env.PLAYWRIGHT_PATH || 'playwright');
+const base = process.env.WEBUI_TEST_URL;
+(async () => {
+  const browser = await chromium.launch({ headless: true, executablePath: process.env.CHROMIUM_PATH || '/usr/bin/chromium' });
+  try {
+    const page = await browser.newPage();
+    const errors = []; page.on('pageerror', e => errors.push(e.message));
+    await page.route('https://api.github.com/**', route => route.fulfill({ json: [] }));
+    const requests = []; page.on('request', r => { if (r.url().includes('/api/upload')) requests.push(new URL(r.url()).pathname); });
+    await page.goto(base);
+    await page.waitForFunction(() => !document.querySelector('#browse-files').disabled);
+    // 600 small files, 3 medium, 2 large: made in the page, queued as a person drops them.
+    const summary = await page.evaluate(async () => {
+      const make = (name, size, seed) => { const b = new Uint8Array(size); for (let i = 0; i < size; i += 4096) b[i] = (seed + i) & 255; return new File([b], name); };
+      const files = [];
+      for (let i = 0; i < 600; i++) files.push(make(`shot-${String(i).padStart(4, '0')}.png`, 1000 + (i * 997) % 60000, i));
+      for (let i = 0; i < 3; i++) files.push(make(`medium-${i}.bin`, 9 * 1024 * 1024 + i, 7 + i));
+      for (let i = 0; i < 2; i++) files.push(make(`large-${i}.iso`, 150 * 1024 * 1024 + 4097 * i, 31 + i));
+      const started = performance.now();
+      queueFiles(files, 'engine');
+      await new Promise(resolve => { const t = setInterval(() => { if (!transfers.some(x => ['queued', 'uploading'].includes(x.state))) { clearInterval(t); resolve(); } }, 50); });
+      const states = {}; for (const x of transfers) states[x.state] = (states[x.state] || 0) + 1;
+      return { states, seconds: (performance.now() - started) / 1000, failed: transfers.filter(x => x.state === 'failed').map(x => x.name + ': ' + x.message).slice(0, 5) };
+    });
+    assert.deepEqual(summary.states, { complete: 605 }, JSON.stringify(summary));
+    const count = name => requests.filter(p => p === name).length;
+    assert.ok(count('/api/upload/batch') >= 3 && count('/api/upload/batch') <= 10, `batches: ${count('/api/upload/batch')}`);
+    assert.equal(count('/api/upload'), 3);
+    assert.equal(count('/api/upload/session'), 2);
+    assert.equal(count('/api/upload/commit'), 2);
+    assert.ok(count('/api/upload/part') >= 12, `parts: ${count('/api/upload/part')}`);
+    // The same names again: every one is refused, none replaced.
+    const again = await page.evaluate(async () => {
+      const files = [new File([new Uint8Array(10)], 'shot-0001.png'), new File([new Uint8Array(9 * 1024 * 1024)], 'medium-0.bin')];
+      const before = transfers.length; queueFiles(files, 'engine');
+      await new Promise(resolve => { const t = setInterval(() => { if (!transfers.some(x => ['queued', 'uploading'].includes(x.state))) { clearInterval(t); resolve(); } }, 50); });
+      return transfers.slice(before).map(x => x.state + ': ' + x.message);
+    });
+    assert.deepEqual(again.map(x => x.split(':')[0]), ['failed', 'failed']);
+    assert.match(again[0], /already exists/); assert.match(again[1], /already exists/);
+    // A large file cancelled mid-way: the session is dropped.
+    const cancelled = await page.evaluate(async () => {
+      const before = transfers.length; queueFiles([new File([new Uint8Array(200 * 1024 * 1024)], 'cancel.iso')], 'engine');
+      const transfer = transfers[before];
+      await new Promise(resolve => { const t = setInterval(() => { if ((transfer.sent || 0) > 20 * 1024 * 1024) { clearInterval(t); resolve(); } }, 10); });
+      cancelTransfer(transfer);
+      await new Promise(resolve => setTimeout(resolve, 1500));
+      return transfer.state;
+    });
+    assert.equal(cancelled, 'cancelled');
+    assert.deepEqual(errors, []);
+    console.log(`PASS: 605 files (600 small in ${count('/api/upload/batch')} batches, 3 single, 2 large in ${count('/api/upload/part')} parts) in ${summary.seconds.toFixed(1)} s; duplicates refused; cancel drops the session.`);
+  } finally { await browser.close(); }
+})().catch(error => { console.error(error); process.exit(1); });

tests/webui_update_fixture.py

@@ -14,7 +14,7 @@ shutil.copytree('webui',r/'webui',dirs_exist_ok=True);shutil.copytree('build/web
 (r/'config/retroarch.cfg').write_text('audio_volume = "0"\ninput_rumble_gain = "100"\n')
 for file in Path('dist/PPSA99169/cores').glob('*.so'):(r/'cores'/file.name).touch()
 shutil.copytree('dist/PPSA99169/info',r/'info',dirs_exist_ok=True)
-subprocess.run(['c++','-std=c++17','-pthread','-I.deps/webui/libmicrohttpd-1.0.10/src/include','-I.deps/native/zlib/zlib-1.3.2/contrib/minizip','-Ivendor/retroarch/deps/mbedtls','tests/webui_server_main.cpp','src/webui_ps5.cpp','src/webui_update.cpp','build/webui-mhd-host/src/microhttpd/.libs/libmicrohttpd.a','build/webui-update-host/libupdate.a','-lz','-o',str(r/'server')],check=True)
+subprocess.run(['c++','-std=c++17','-pthread','-I.deps/webui/libmicrohttpd-1.0.10/src/include','-I.deps/native/zlib/zlib-1.3.2/contrib/minizip','-Ivendor/retroarch/deps/mbedtls','tests/webui_server_main.cpp','src/webui_ps5.cpp','src/webui_transfer.cpp','src/webui_update.cpp','build/webui-mhd-host/src/microhttpd/.libs/libmicrohttpd.a','build/webui-update-host/libupdate.a','-lz','-o',str(r/'server')],check=True)
 try:
     subprocess.run([str(r/'server'),str(r),'6770'],check=True)
 finally:

tools/build-webui-daemon.sh

@@ -15,7 +15,7 @@ includes=(-I"$root/src" -I"$root/.deps/webui/libmicrohttpd-1.0.10/src/include"
     -I"$root/.deps/native/zlib/zlib-1.3.2" -I"$root/.deps/native/zlib/zlib-1.3.2/contrib/minizip"
     -I"$root/vendor/retroarch/deps/mbedtls")
 objects=()
-for source in daemon/webui_daemon src/webui_ps5 src/webui_update; do
+for source in daemon/webui_daemon src/webui_ps5 src/webui_transfer src/webui_update; do
     "$sdk/bin/prospero-clang++" -std=c++17 -O2 -Wall -Werror "${includes[@]}" \
         -c "$root/$source.cpp" -o "$build/${source##*/}.o"
     objects+=("$build/${source##*/}.o")

webui/app.js

@@ -197,7 +197,7 @@ function drawTransfers() {
           progress = document.createElement('progress'); progress.max = 100;
           progress.setAttribute('aria-label', `Upload progress for ${transfer.name}`); details.append(progress);
           const cancel = element('button', 'Cancel', 'secondary');
-          cancel.addEventListener('click', () => { transfer.state = 'cancelled'; transfer.message = 'Cancelled'; if (transfer.xhr) transfer.xhr.abort(); transfer.file = null; drawTransfers(); });
+          cancel.addEventListener('click', () => cancelTransfer(transfer));
           row.append(cancel);
         }
         progress.value = transfer.percent || 0;
@@ -216,35 +216,175 @@ function queueFiles(files, destination) {
   }
   drawTransfers(); sendNext();
 }
+// The transfer engine (src/webui_transfer.h on the console): six lanes run at once.
+// Small files travel many to a request (a batch), large files as parts on every free
+// lane (an upload session), the rest one to a request. Every file is renamed into
+// place on the console only once whole.
+const LANES = 6, SMALL = 4 * 1024 ** 2, LARGE = 64 * 1024 ** 2, PART = 32 * 1024 ** 2;
+const BATCH_FILES = 256, BATCH_BYTES = 16 * 1024 ** 2, PART_TRIES = 3;
+let jobs = [], opening = 0; // sessions being opened: their parts are about to queue
+function cancelTransfer(transfer) {
+  if (!['queued', 'uploading'].includes(transfer.state)) return;
+  transfer.state = 'cancelled'; transfer.message = 'Cancelled';
+  for (const xhr of transfer.requests || []) xhr.abort();
+  if (transfer.session) api('/api/upload/session?id=' + encodeURIComponent(transfer.session), { method: 'DELETE' }).catch(() => {});
+  transfer.file = null; drawTransfers();
+}
+function showProgress(transfer, sent) {
+  transfer.sent = Math.min(sent, transfer.file?.size ?? sent);
+  const size = transfer.file?.size || 0, percent = size ? Math.floor(transfer.sent / size * 100) : 100;
+  if (percent === transfer.percent && transfer.state === 'uploading') return;
+  transfer.percent = percent; transfer.state = 'uploading';
+  transfer.message = percent >= 100 ? 'Finishing on the console…' : `${percent}% · ${bytes(transfer.sent)} of ${bytes(size)}`;
+  const now = Date.now(); if (now - (transfer.drawn || 0) > 150 || percent >= 100) { transfer.drawn = now; drawTransfers(); }
+}
+function finish(transfer, state, message) {
+  if (transfer.state === 'cancelled') return;
+  transfer.state = state; transfer.message = message; transfer.file = null; transfer.requests = [];
+  transfer.session = null; drawTransfers();
+}
+// One request on a lane: resolves with {status, body}, or rejects when the connection fails.
+function send(method, url, body, transfers, onProgress) {
+  return new Promise((resolve, reject) => {
+    const xhr = new XMLHttpRequest();
+    for (const transfer of transfers) (transfer.requests ||= []).push(xhr);
+    const done = () => { for (const transfer of transfers) transfer.requests = (transfer.requests || []).filter(x => x !== xhr); };
+    xhr.open(method, url); xhr.setRequestHeader('X-RetroArch-Token', token);
+    if (onProgress) xhr.upload.onprogress = event => onProgress(event.loaded);
+    xhr.onload = () => { done(); let parsed = {}; try { parsed = JSON.parse(xhr.responseText); } catch {} resolve({ status: xhr.status, body: parsed }); };
+    xhr.onerror = () => { done(); reject(new Error('Connection lost. Reconnect, then choose this file again.')); };
+    xhr.onabort = () => { done(); reject(new Error('Cancelled')); };
+    xhr.send(body);
+  });
+}
+// The batch stream: per file a u16 name length, the name (UTF-8), a u64 size and the
+// file itself, then a zero length. A Blob of the files' own Blobs: nothing is copied.
+function batchBody(transfers) {
+  const parts = [], encoder = new TextEncoder();
+  for (const transfer of transfers) {
+    const name = encoder.encode(transfer.path), head = new Uint8Array(2 + name.length + 8), view = new DataView(head.buffer);
+    view.setUint16(0, name.length, true); head.set(name, 2); view.setBigUint64(2 + name.length, BigInt(transfer.file.size), true);
+    parts.push(head, transfer.file);
+  }
+  parts.push(new Uint8Array(2));
+  return new Blob(parts);
+}
+async function runBatch(transfers) {
+  transfers = transfers.filter(transfer => transfer.state === 'queued');
+  if (!transfers.length) return;
+  const encoder = new TextEncoder(), starts = []; let at = 0;
+  for (const transfer of transfers) { at += 2 + encoder.encode(transfer.path).length + 8; starts.push(at); at += transfer.file.size; }
+  for (const transfer of transfers) showProgress(transfer, 0);
+  try {
+    const { status, body: result } = await send('PUT', '/api/upload/batch', batchBody(transfers), transfers,
+      loaded => transfers.forEach((transfer, i) => showProgress(transfer, loaded - starts[i])));
+    const failed = new Map((result.failed || []).map(entry => [entry.name, entry.error]));
+    const skipped = new Set(result.skippedNames || []);
+    for (const transfer of transfers) {
+      if (transfer.state !== 'uploading') continue;
+      if (skipped.has(transfer.path)) finish(transfer, 'failed', 'A file with this name already exists. Rename your file first.');
+      else if (failed.has(transfer.path)) finish(transfer, 'failed', `Upload failed (${failed.get(transfer.path)}). Choose the file again to retry.`);
+      else if (status === 200) finish(transfer, 'complete', `Uploaded · ${bytes(transfer.file.size)}`);
+      else finish(transfer, 'failed', result.error || result.problem || 'Upload failed. Choose the file again to retry.');
+    }
+  } catch (error) {
+    for (const transfer of transfers) if (transfer.state === 'uploading') finish(transfer, 'failed', error.message);
+    if (error.message !== 'Cancelled') setConnection(false);
+  }
+}
+async function runSingle(transfer) {
+  if (transfer.state !== 'queued') return;
+  showProgress(transfer, 0);
+  try {
+    const { status, body } = await send('PUT', '/api/upload?path=' + encodeURIComponent(transfer.path), transfer.file, [transfer], loaded => showProgress(transfer, loaded));
+    if (status === 201) finish(transfer, 'complete', `Uploaded · ${bytes(transfer.file.size)}`);
+    else finish(transfer, 'failed', body.error || 'Upload failed. Choose the file again to retry.');
+  } catch (error) { finish(transfer, error.message === 'Cancelled' ? 'cancelled' : 'failed', error.message); if (error.message !== 'Cancelled') setConnection(false); }
+}
+// A large file: a session, then its parts as jobs any lane takes. The part that
+// completes the file commits it, on its own lane: no lane ever waits on another.
+async function runLarge(transfer) {
+  if (transfer.state !== 'queued') return;
+  showProgress(transfer, 0);
+  let opened;
+  opening++;
+  try { opened = await send('POST', `/api/upload/session?path=${encodeURIComponent(transfer.path)}&size=${transfer.file.size}`, null, [transfer]); }
+  catch (error) { finish(transfer, 'failed', error.message); return; }
+  finally { opening--; }
+  if (opened.status !== 201) { finish(transfer, 'failed', opened.body.error || 'Upload failed. Choose the file again to retry.'); return; }
+  transfer.session = opened.body.id;
+  const size = transfer.file.size, part = Math.max(4 * 1024 ** 2, Math.min(opened.body.partSize || PART, Math.ceil(size / LANES)));
+  const sent = new Map(), parts = [];
+  for (let start = 0; start < size; start += part) parts.push([start, Math.min(start + part, size)]);
+  let left = parts.length, failure = null;
+  const abandon = message => {
+    if (failure) return;
+    failure = message;
+    api('/api/upload/session?id=' + encodeURIComponent(transfer.session), { method: 'DELETE' }).catch(() => {});
+    if (message !== 'Cancelled') finish(transfer, 'failed', message);
+  };
+  const commit = async () => {
+    try {
+      const { status, body } = await send('POST', '/api/upload/commit?id=' + encodeURIComponent(transfer.session), null, [transfer]);
+      if (status === 201) finish(transfer, 'complete', `Uploaded · ${bytes(size)}`);
+      else finish(transfer, 'failed', body.error || 'Upload failed. Choose the file again to retry.');
+    } catch (error) { finish(transfer, 'failed', error.message); }
+  };
+  const partJob = ([start, end]) => async () => {
+    for (let attempt = 1; !failure; attempt++) {
+      if (transfer.state !== 'uploading') { abandon('Cancelled'); return; }
+      try {
+        const { status, body } = await send('PUT', `/api/upload/part?id=${encodeURIComponent(transfer.session)}&offset=${start}`, transfer.file.slice(start, end), [transfer],
+          loaded => { sent.set(start, loaded); showProgress(transfer, [...sent.values()].reduce((a, b) => a + b, 0)); });
+        if (status === 200) { sent.set(start, end - start); break; }
+        if (attempt >= PART_TRIES || status === 404 || status === 416) { abandon(body.error || 'A part could not be written.'); return; }
+      } catch (error) {
+        if (error.message === 'Cancelled') { abandon('Cancelled'); return; }
+        if (attempt >= PART_TRIES) { abandon(error.message); return; }
+      }
+    }
+    if (!failure && --left === 0) await commit();
+  };
+  // First in the queue, so every free lane joins this file at once.
+  jobs.unshift(...parts.map(partJob));
+}
+// Turns the queued transfers into jobs: batches of small files, sessions, single files.
+function planJobs() {
+  let batch = [], batchBytes = 0;
+  const flush = () => { if (batch.length) { const group = batch; jobs.push(() => runBatch(group)); } batch = []; batchBytes = 0; };
+  for (const transfer of transfers) {
+    if (transfer.state !== 'queued' || transfer.planned) continue;
+    transfer.planned = true;
+    const size = transfer.file.size;
+    if (size < SMALL) {
+      if (batch.length >= BATCH_FILES || batchBytes + size > BATCH_BYTES) flush();
+      batch.push(transfer); batchBytes += size;
+    } else if (size >= LARGE) jobs.push(() => runLarge(transfer));
+    else jobs.push(() => runSingle(transfer));
+  }
+  flush();
+}
 async function sendNext() {
+  planJobs();
   if (sending) return;
   sending = true;
   try {
-    for (const transfer of transfers) {
-      if (transfer.state !== 'queued') continue;
-      if (!connected && !await reconnect()) { transfer.state = 'failed'; transfer.message = 'Console disconnected. Reconnect, then choose this file again.'; transfer.file = null; drawTransfers(); continue; }
-      await new Promise(resolve => {
-        const xhr = new XMLHttpRequest(); transfer.xhr = xhr; transfer.state = 'uploading';
-        transfer.message = 'Uploading…'; drawTransfers();
-        xhr.open('PUT', '/api/upload?path=' + encodeURIComponent(transfer.path));
-        xhr.setRequestHeader('X-RetroArch-Token', token);
-        let lastProgress = 0;
-        xhr.upload.onprogress = event => {
-          if (!event.lengthComputable) return;
-          transfer.percent = Math.floor(event.loaded / event.total * 100);
-          transfer.message = transfer.percent === 100 ? 'Finishing on the console…' : `${transfer.percent}% · ${bytes(event.loaded)} of ${bytes(event.total)}`;
-          if (Date.now() - lastProgress > 150 || transfer.percent === 100) { lastProgress = Date.now(); drawTransfers(); }
-        };
-        xhr.onload = () => {
-          if (xhr.status === 201) { transfer.state = 'complete'; transfer.message = `Uploaded · ${bytes(transfer.file.size)}`; }
-          else { transfer.state = 'failed'; try { transfer.message = JSON.parse(xhr.responseText).error; } catch { transfer.message = 'Upload failed. Choose the file again to retry.'; } }
-        };
-        xhr.onerror = () => { transfer.state = 'failed'; transfer.message = 'Connection lost. Reconnect, then choose this file again.'; setConnection(false); };
-        xhr.onabort = () => { transfer.state = 'cancelled'; transfer.message = 'Cancelled'; };
-        xhr.onloadend = () => { transfer.file = null; transfer.xhr = null; drawTransfers(); resolve(); };
-        xhr.send(transfer.file);
-      });
+    if (!connected && !await reconnect()) {
+      for (const transfer of transfers) if (transfer.state === 'queued') finish(transfer, 'failed', 'Console disconnected. Reconnect, then choose this file again.');
+      jobs = []; return;
     }
+    // Each lane runs one job at a time until none is left; a job is one request (or a
+    // part and its file's commit), so no lane waits on another lane's work.
+    const lane = async () => {
+      for (;;) {
+        planJobs();
+        const job = jobs.shift();
+        if (job) await job();
+        else if (opening) await new Promise(resolve => setTimeout(resolve, 25));
+        else return;
+      }
+    };
+    await Promise.all(Array.from({ length: LANES }, lane));
   } finally { sending = false; await Promise.all([loadLibrary(), loadContent(currentPath), reconnect()]); }
 }
 let pickerDestination = '';