diff --git a/cmake/Sodium.cmake b/cmake/Sodium.cmake
index dbdc5d88..1597ee9a 100644
--- a/cmake/Sodium.cmake
+++ b/cmake/Sodium.cmake
@@ -11,7 +11,9 @@ file(GLOB_RECURSE SODIUM_C "${libsodium_SOURCE_DIR}/src/*.c")
 file(GLOB_RECURSE SODIUM_H "${libsodium_SOURCE_DIR}/src/*.h")
 
 add_library(sodium STATIC ${SODIUM_C} ${SODIUM_H})
-
+if(CMAKE_C_COMPILER_ID MATCHES "GNU|Clang")
+    target_compile_options(sodium PRIVATE -w)
+endif()
 target_include_directories(sodium PUBLIC $<BUILD_INTERFACE:${libsodium_SOURCE_DIR}/src/libsodium/include> $<INSTALL_INTERFACE:include/opencmw> PRIVATE "${libsodium_SOURCE_DIR}/src/libsodium/include/sodium")
 # silence warnings about being built by a different build system
 target_compile_definitions(sodium PRIVATE CONFIGURED)
diff --git a/docs/RestUriMapping.md b/docs/RestUriMapping.md
index d185ea4e..d05937d4 100644
--- a/docs/RestUriMapping.md
+++ b/docs/RestUriMapping.md
@@ -18,10 +18,69 @@ The REST backend maps HTTP requests to Majordomo requests.
 By default, `PUT` and `POST` HTTP requests are mapped to Majordomo's `Set`,
 and `GET` HTTP request is mapped to Majordomo's `Get` request.
 
-Requests with a query parameter `LongPollingIdx` are treated as `LongPoll` request subscribing to the
-given topic (consisting of URI path and other query parameters), with the possible values:
+Use `LongPollingIdx` on a GET request to read notifications for a topic
+(the path and other query parameters):
 
- - `Next`: Redirects to the next notification message that arrives after the request has been processed.
- - `Last`: Redirects to the most recent notification message that is in the cache when the request is
- processed. If there is no such entry yet, it's treated like `Next`, i.e. waits for the next notification.
- - a positive integer value, to retrieve a specific cache entry.
+- `Next`: Redirect to the next new message's index.
+- `Last`: Redirect to the newest buffered message, or use `Next` if the buffer is empty.
+- A non-negative integer: Read that message, waiting if it has not arrived.
+
+Add `LongPollingBatch` to receive several messages in one HTTP response:
+
+- `?LongPollingIdx=42&LongPollingBatch=10`:
+  Wait for messages 42 through 51, then return all ten.
+- `?LongPollingIdx=42&LongPollingBatch=AllAvailable`:
+  Return message 42 and all newer buffered messages. If 42 has not arrived, wait for it.
+- `?LongPollingIdx=42` (no batch):
+  Return only message 42, waiting if needed.
+
+The server keeps the latest 100 messages per subscription. If the requested start is too old, a batch starts
+at the oldest buffered message. A numeric batch still waits for exactly the requested count.
+Without a batch, a too-old index redirects to the next message, as before.
+
+Clients report detected gaps in `Message::error`, alongside the next valid sample.
+The warning does not invalidate that sample or stop the subscription.
+
+`LongPollingBatch` requires `LongPollingIdx`. The batch size must be 1–100 or `AllAvailable`;
+it does not mean "the latest N messages" or a time range. Invalid values and index ranges too
+large to represent return HTTP `400`.
+
+Batches are intended for direct connections. Proxy caching is unchanged.
+
+If a successful batch response cannot be decoded, the client reports an error and continues after its
+last index, if known; otherwise it uses `Next`, which may skip buffered messages.
+HTTP `504` retries the same index if known. Other HTTP errors stop the batch subscription.
+
+Batch responses use `multipart/mixed`, with one part per message. Each part carries its own
+index, topic, service name, and payload length. The outer `x-opencmw-long-polling-idx` header
+gives the first returned index; topic and service name appear only in the parts.
+
+Example with two messages. All line breaks are `\r\n`, including after the closing boundary:
+
+```text
+--opencmw-long-polling-multipart-boundary
+x-opencmw-long-polling-idx: 42
+x-opencmw-topic: /colors?sample=42
+x-opencmw-service-name: colors-service
+content-length: 5
+
+hello
+--opencmw-long-polling-multipart-boundary
+x-opencmw-long-polling-idx: 43
+x-opencmw-topic: /colors?sample=43
+x-opencmw-service-name: colors-service
+content-length: 5
+
+world
+--opencmw-long-polling-multipart-boundary--
+```
+
+Response header:
+`content-type: multipart/mixed; boundary=opencmw-long-polling-multipart-boundary`.
+
+The boundary is fixed. Read each payload by its `content-length`, not by searching for the
+boundary. The length counts payload bytes only, excluding the following line break.
+
+Generic MIME parsers can also be used, but they find parts by searching for boundary delimiter
+lines rather than using `content-length`. We do not guarantee that the boundary is absent from
+payloads. A matching delimiter line inside a payload may therefore be mistaken for the end of a part.
diff --git a/src/client/include/RestClientEmscripten.hpp b/src/client/include/RestClientEmscripten.hpp
index 74b2dfdf..81157368 100644
--- a/src/client/include/RestClientEmscripten.hpp
+++ b/src/client/include/RestClientEmscripten.hpp
@@ -14,6 +14,7 @@
 #include <concepts>
 #include <cstdint>
 #include <cstdio>
+#include <expected>
 #include <format>
 #include <functional>
 #include <iostream>
@@ -34,7 +35,9 @@
 
 #include <ClientCommon.hpp>
 #include <ClientContext.hpp>
+#include <LongPollingBatch.hpp>
 #include <MIME.hpp>
+#include <Topic.hpp>
 #include <URI.hpp>
 
 using namespace opencmw;
@@ -72,6 +75,49 @@ inline std::optional<std::uint64_t> parseLongPollingIndex(std::string_view respo
     }
 }
 
+inline bool usesLongPollingBatch(const mdp::Message::URI &topic) {
+    return topic.queryParamMap().contains(std::string(long_polling::kBatchParameter));
+}
+
+inline std::expected<std::string, std::string> takeInitialLongPollingIndex(Command &command) {
+    auto       parameters = command.topic.queryParamMap();
+    const auto entry      = parameters.find(std::string(long_polling::kIndexParameter));
+    if (entry == parameters.end()) {
+        return "Next";
+    }
+    if (!entry->second.has_value()) {
+        return std::unexpected("LongPollingIdx requires a value");
+    }
+
+    std::string index = *entry->second;
+    if (index != "Next" && index != "Last") {
+        auto parsed = long_polling::parseUnsigned(*entry->second, long_polling::kIndexParameter);
+        if (!parsed.has_value()) {
+            return std::unexpected(parsed.error());
+        }
+        index = std::to_string(*parsed);
+    }
+
+    parameters.erase(entry);
+    command.topic = mdp::Message::URI::UriFactory(command.topic).setQuery(std::move(parameters)).build();
+    return index;
+}
+
+inline mdp::Topic subscriptionTopic(const mdp::Message::URI &topic) {
+    auto parameters = topic.queryParamMap();
+    parameters.erase(std::string(long_polling::kIndexParameter));
+    parameters.erase(std::string(long_polling::kBatchParameter));
+    const auto normalized = mdp::Message::URI::UriFactory(topic).setQuery(std::move(parameters)).build();
+    return mdp::Topic::fromMdpTopic(normalized);
+}
+
+inline bool sameSubscription(const mdp::Message::URI &lhs, const mdp::Message::URI &rhs) {
+    return lhs.scheme() == rhs.scheme()
+        && lhs.hostName() == rhs.hostName()
+        && lhs.port() == rhs.port()
+        && subscriptionTopic(lhs) == subscriptionTopic(rhs);
+}
+
 struct SubscriptionState {
     Command                      command{};
     std::optional<std::uint64_t> lastDeliveredIndex{};
@@ -129,14 +175,30 @@ struct RestWorkerState {
     }
 
     void startSubscription(Command &&cmd) {
-        const std::uint64_t id = _nextSubscriptionId++;
-        _subscriptions.emplace(id, SubscriptionState{ .command = std::move(cmd) });
-        startNextLongPoll(id, std::nullopt);
+        const auto initialIndex = takeInitialLongPollingIndex(cmd);
+        if (!initialIndex.has_value()) {
+            reportFailure(cmd, initialIndex.error());
+            return;
+        }
+        const bool usesBatch = usesLongPollingBatch(cmd.topic);
+        if (std::ranges::any_of(_subscriptions, [&](const auto &entry) {
+                return sameSubscription(entry.second.command.topic, cmd.topic)
+                    && (usesBatch || usesLongPollingBatch(entry.second.command.topic));
+            })) {
+            reportFailure(cmd, "A subscription for this topic is already active; concurrent batch subscriptions are not supported");
+            return;
+        }
+        const std::uint64_t id           = _nextSubscriptionId++;
+        auto              &subscription = _subscriptions.emplace(id, SubscriptionState{ .command = std::move(cmd) }).first->second;
+        if (const auto index = long_polling::parseUnsigned(*initialIndex, long_polling::kIndexParameter); index.has_value() && *index > 0) {
+            subscription.lastDeliveredIndex = *index - 1; // Detect a gap before the first returned sample.
+        }
+        startLongPoll(id, *initialIndex);
     }
 
     void stopSubscription(const Command &cmd) {
         const auto entry = std::ranges::find_if(_subscriptions,
-                [&](const auto &pair) { return pair.second.command.topic == cmd.topic; });
+                [&](const auto &pair) { return sameSubscription(pair.second.command.topic, cmd.topic); });
         if (entry == _subscriptions.end()) {
             return;
         }
@@ -147,7 +209,7 @@ struct RestWorkerState {
         }
     }
 
-    void startNextLongPoll(std::uint64_t subscriptionId, std::optional<std::uint64_t> index) noexcept {
+    void startLongPoll(std::uint64_t subscriptionId, std::string longPollingIndex) noexcept {
         try {
             if (!_acceptWork.load(std::memory_order_acquire)) {
                 return;
@@ -156,13 +218,12 @@ struct RestWorkerState {
             if (entry == _subscriptions.end()) {
                 return;
             }
-            const std::string longPollingIndex = index.has_value() ? std::to_string(*index) : "Next";
 
-            auto              activeFetch      = std::make_unique<ActiveFetch>();
-            activeFetch->owner                 = this;
-            activeFetch->id                    = _nextFetchId++;
-            activeFetch->subscriptionId        = subscriptionId;
-            entry->second.activeFetchId        = activeFetch->id;
+            auto activeFetch            = std::make_unique<ActiveFetch>();
+            activeFetch->owner          = this;
+            activeFetch->id             = _nextFetchId++;
+            activeFetch->subscriptionId = subscriptionId;
+            entry->second.activeFetchId = activeFetch->id;
             startFetch(std::move(activeFetch), URI<STRICT>::UriFactory(entry->second.command.topic).addQueryParameter("LongPollingIdx", longPollingIndex).build());
         } catch (const std::exception &e) {
             endSubscription(subscriptionId, nullptr, 500, {}, e.what());
@@ -171,6 +232,10 @@ struct RestWorkerState {
         }
     }
 
+    void startNextLongPoll(std::uint64_t subscriptionId, std::optional<std::uint64_t> index) noexcept {
+        startLongPoll(subscriptionId, index.has_value() ? std::to_string(*index) : "Next");
+    }
+
     void startGetOrSet(Command &&cmd) {
         const URI<STRICT> uri         = cmd.topic;
 
@@ -302,6 +367,50 @@ struct RestWorkerState {
             return;
         }
 
+        if (usesLongPollingBatch(state.command.topic)) {
+            const auto parts = long_polling::decodeBatch(body);
+            if (!parts.has_value()) {
+                resumeSubscriptionAfterBatchError(fetchId, subscriptionId, fetch, std::format("could not decode long-polling batch: {}", parts.error()));
+                return;
+            }
+
+            auto                      lastDelivered = state.lastDeliveredIndex;
+            std::vector<mdp::Message> messages;
+            messages.reserve(parts->size());
+            try {
+                for (const auto &part : *parts) {
+                    if (lastDelivered.has_value() && part.index <= *lastDelivered) {
+                        continue;
+                    }
+                    std::string skippedWarning;
+                    if (lastDelivered.has_value() && part.index - *lastDelivered > 1) {
+                        skippedWarning = std::format("Warning: skipped {} samples", part.index - *lastDelivered - 1);
+                    }
+                    auto message        = buildMessage(state.command, status, part.payload, skippedWarning);
+                    message.topic       = mdp::Message::URI(std::string(part.topic));
+                    message.serviceName = std::string(part.serviceName);
+                    messages.push_back(std::move(message));
+                    lastDelivered = part.index;
+                }
+            } catch (const std::exception &e) {
+                const auto lastIndex = parts->back().index;
+                resumeSubscriptionAfterBatchError(fetchId, subscriptionId, fetch, std::format("could not parse long-polling batch metadata: {}", e.what()),
+                        lastIndex < std::numeric_limits<std::uint64_t>::max() ? std::optional{ lastIndex + 1 } : std::nullopt);
+                return;
+            }
+
+            state.lastDeliveredIndex = lastDelivered;
+            const auto callback      = state.command.callback;
+            closeFetch(fetchId, fetch);
+            for (const auto &message : messages) {
+                invokeGuarded(callback, message);
+            }
+            if (lastDelivered.has_value()) {
+                startNextLongPoll(subscriptionId, *lastDelivered + 1);
+            }
+            return;
+        }
+
         if (!index.has_value()) {
             endSubscription(subscriptionId, fetch, status, body, "missing or unparsable LongPollingIdx in the response URL");
             return;
@@ -327,6 +436,14 @@ struct RestWorkerState {
         startNextLongPoll(subscriptionId, *index + 1);
     }
 
+    void resumeSubscriptionAfterBatchError(std::uint64_t fetchId, std::uint64_t subscriptionId, emscripten_fetch_t *fetch, std::string_view error, std::optional<std::uint64_t> nextIndex = std::nullopt) {
+        const auto command = _subscriptions.at(subscriptionId).command;
+        const auto status  = fetch->status;
+        closeFetch(fetchId, fetch);
+        reportFailure(command, error, status);
+        startNextLongPoll(subscriptionId, nextIndex);
+    }
+
     void handleGetOrSetCompletion(std::uint64_t fetchId, emscripten_fetch_t *fetch, std::optional<std::string_view> fetchError) {
         if (!_acceptWork.load(std::memory_order_acquire)) {
             closeFetch(fetchId, fetch);
@@ -408,13 +525,13 @@ struct RestWorkerState {
         emscripten_runtime_keepalive_pop();
     }
 
-    void        reportFailure(const Command &command, std::string_view error) noexcept {
+    void reportFailure(const Command &command, std::string_view error, unsigned short status = 500) noexcept {
         if (!command.callback) {
             std::println(std::cerr, "RestClientEmscripten: {}", error);
             return;
         }
         try {
-            invokeGuarded(command.callback, buildMessage(command, 500, {}, error));
+            invokeGuarded(command.callback, buildMessage(command, status, {}, error));
         } catch (const std::exception &e) {
             std::println(std::cerr, "RestClientEmscripten: could not report '{}': {}", error, e.what());
         } catch (...) {
diff --git a/src/client/include/RestClientNative.hpp b/src/client/include/RestClientNative.hpp
index 0707e209..aa367297 100644
--- a/src/client/include/RestClientNative.hpp
+++ b/src/client/include/RestClientNative.hpp
@@ -5,8 +5,10 @@
 #include <chrono>
 #include <cstring>
 #include <deque>
+#include <expected>
 #include <fcntl.h>
 #include <format>
+#include <limits>
 #include <mutex>
 #include <netinet/tcp.h>
 #include <poll.h>
@@ -15,12 +17,14 @@
 #include <string_view>
 #include <sys/poll.h>
 #include <thread>
+#include <tuple>
 #include <unistd.h>
 
 #include <nghttp2/nghttp2.h>
 
 #include "ClientCommon.hpp"
 #include "ClientContext.hpp"
+#include "LongPollingBatch.hpp"
 #include "MdpMessage.hpp"
 #include "MIME.hpp"
 #include "rest/RestUtils.hpp"
@@ -39,6 +43,46 @@ namespace detail {
 
 using namespace opencmw::rest::detail;
 
+inline bool usesLongPollingBatch(const mdp::Message::URI &topic) {
+    return topic.queryParamMap().contains(std::string(long_polling::kBatchParameter));
+}
+
+inline std::string subscriptionKey(const mdp::Message::URI &topic) {
+    auto parameters = topic.queryParamMap();
+    parameters.erase(std::string(long_polling::kIndexParameter));
+    parameters.erase(std::string(long_polling::kBatchParameter));
+    const auto subscriptionTopic = mdp::Message::URI::UriFactory(topic).setQuery(std::move(parameters)).build();
+    return mdp::Topic::fromMdpTopic(subscriptionTopic).toZmqTopic();
+}
+
+inline std::expected<std::optional<std::uint64_t>, std::string> takeInitialLongPollingIndex(Command &command, SubscriptionMode &mode) {
+    auto       parameters = command.topic.queryParamMap();
+    const auto entry      = parameters.find(std::string(long_polling::kIndexParameter));
+    if (entry == parameters.end()) {
+        return std::optional<std::uint64_t>{};
+    }
+    if (!entry->second.has_value()) {
+        return std::unexpected("LongPollingIdx requires a value");
+    }
+
+    std::optional<std::uint64_t> index;
+    if (*entry->second == "Next") {
+        mode = SubscriptionMode::Next;
+    } else if (*entry->second == "Last") {
+        mode = SubscriptionMode::Last;
+    } else {
+        auto parsed = long_polling::parseUnsigned(*entry->second, long_polling::kIndexParameter);
+        if (!parsed.has_value()) {
+            return std::unexpected(parsed.error());
+        }
+        index = *parsed;
+    }
+
+    parameters.erase(entry);
+    command.topic = mdp::Message::URI::UriFactory(command.topic).setQuery(std::move(parameters)).build();
+    return index;
+}
+
 template<typename T>
 struct SharedQueue {
     // TODO use a lock-free queue? This is only used client-side, so not that critical
@@ -217,7 +261,9 @@ struct ClientSessionBase {
         RequestResponse rr;
         rr.request = std::move(cmd);
         try {
-            rr.normalizedTopic = mdp::Topic::fromMdpTopic(rr.request.topic).toZmqTopic();
+            rr.normalizedTopic = rr.request.command == mdp::Command::Subscribe
+                                       ? subscriptionKey(rr.request.topic)
+                                       : mdp::Topic::fromMdpTopic(rr.request.topic).toZmqTopic();
         } catch (...) {
             rr.normalizedTopic = rr.request.topic.str();
         }
@@ -245,6 +291,7 @@ struct ClientSessionBase {
                 std::optional<URI<>> location;
                 try {
                     location = URI<>(it->second.location);
+                    std::ignore = location->queryParamMap(); // Lazy query parsing here so errors are caught.
                 } catch (const std::exception &e) {
                     HTTP_DBG("Client::Header: Could not parse URI '{}': {}", it->second.location, e.what());
                     it->second.reportError(std::format("Could not parse redirect URI '{}': {}", it->second.location, e.what()));
@@ -277,12 +324,29 @@ struct ClientSessionBase {
                 const auto hasError = !it->second.responseStatus.starts_with("2") && !it->second.responseStatus.starts_with("3");
                 if (hasError) {
                     response.error = std::move(it->second.payload);
+                } else if (request.command == mdp::Command::Subscribe && usesLongPollingBatch(request.topic)) {
+                    const auto parts = long_polling::decodeBatch(it->second.payload);
+                    if (!parts.has_value()) {
+                        response.error = std::format("Could not decode long-polling batch: {}", parts.error());
+                        resumeSubscriptionAfterBatchError(it->second.normalizedTopic, std::move(response));
+                    } else {
+                        handleSubscriptionBatchResponse(it->second.normalizedTopic, *parts, std::move(response));
+                    }
+                    _requestsByStreamId.erase(it);
+                    return 0;
                 } else {
                     response.data = IoBuffer(it->second.payload.data(), it->second.payload.size());
                 }
-                if (it->second.longPollingIdx) {
+                if (request.command == mdp::Command::Subscribe) {
                     // Subscription
-                    handleSubscriptionResponse(it->second.normalizedTopic, it->second.longPollingIdx.value(), std::move(response));
+                    if (hasError && (usesLongPollingBatch(request.topic) || !it->second.longPollingIdx.has_value())) {
+                        reportSubscriptionError(it->second.normalizedTopic, std::move(response));
+                    } else if (!it->second.longPollingIdx.has_value()) {
+                        response.error = "Long-polling response is missing x-opencmw-long-polling-idx";
+                        reportSubscriptionError(it->second.normalizedTopic, std::move(response));
+                    } else {
+                        handleSubscriptionResponse(it->second.normalizedTopic, it->second.longPollingIdx.value(), std::move(response), hasError);
+                    }
                 } else {
                     // GET/SET
                     if (request.callback) {
@@ -304,19 +368,107 @@ struct ClientSessionBase {
         _requestsByStreamId.clear();
     }
 
-    void handleSubscriptionResponse(std::string zmqTopic, std::uint64_t longPollingIdx, mdp::Message &&response) {
+    static std::string makeSkippedWarning(std::optional<std::uint64_t> lastDelivered, std::uint64_t index) {
+        if (!lastDelivered.has_value() || index - *lastDelivered <= 1) {
+            return {};
+        }
+        return std::format("Warning: skipped {} samples", index - *lastDelivered - 1);
+    }
+
+    void handleSubscriptionResponse(std::string zmqTopic, std::uint64_t longPollingIdx, mdp::Message &&response, bool hasError) {
         auto subIt = _subscriptions.find(zmqTopic);
         if (subIt == _subscriptions.end()) {
             HTTP_DBG("Client::handleSubscriptionResponse: Could not find subscription for topic '{}'", zmqTopic);
             return;
         }
-        auto &sub                      = subIt->second;
+        auto &sub = subIt->second;
+        if (!hasError) {
+            if (sub.lastReceivedLongPollingIdx.has_value() && longPollingIdx <= *sub.lastReceivedLongPollingIdx) {
+                auto repeated = sub.request;
+                submitRequest(std::move(repeated), sub.mode, {}, *sub.lastReceivedLongPollingIdx + 1);
+                return;
+            }
+            response.error = makeSkippedWarning(sub.lastReceivedLongPollingIdx, longPollingIdx);
+        }
+
         sub.lastReceivedLongPollingIdx = longPollingIdx;
         auto request                   = sub.request;
 
         submitRequest(std::move(request), sub.mode, {}, longPollingIdx + kParallelLongPollingRequests);
 
+        invokeSubscriptionCallbacks(sub, std::move(response));
+    }
+
+    void handleSubscriptionBatchResponse(const std::string &zmqTopic, const std::vector<long_polling::BatchPart> &parts, mdp::Message &&response) {
+        auto subIt = _subscriptions.find(zmqTopic);
+        if (subIt == _subscriptions.end()) {
+            HTTP_DBG("Client::handleSubscriptionBatchResponse: Could not find subscription for topic '{}'", zmqTopic);
+            return;
+        }
+        auto                     &sub           = subIt->second;
+        auto                      lastDelivered = sub.lastReceivedLongPollingIdx;
+        std::vector<mdp::Message> messages;
+        messages.reserve(parts.size());
+        try {
+            for (const auto &part : parts) {
+                if (lastDelivered.has_value() && part.index <= *lastDelivered) {
+                    continue; // a redirect or retry re-fetched a range we already delivered
+                }
+                auto partResponse        = response;
+                partResponse.topic       = URI<>(std::string(part.topic));
+                partResponse.serviceName = std::string(part.serviceName);
+                partResponse.data        = IoBuffer(part.payload.data(), part.payload.size());
+                partResponse.error       = makeSkippedWarning(lastDelivered, part.index);
+                messages.push_back(std::move(partResponse));
+                lastDelivered = part.index;
+            }
+        } catch (const std::exception &e) {
+            response.error       = std::format("Could not parse long-polling batch metadata: {}", e.what());
+            const auto lastIndex = parts.back().index;
+            resumeSubscriptionAfterBatchError(zmqTopic, std::move(response),
+                    lastIndex < std::numeric_limits<std::uint64_t>::max() ? std::optional{ lastIndex + 1 } : std::nullopt);
+            return;
+        }
+
+        sub.lastReceivedLongPollingIdx = lastDelivered;
+        auto request                   = sub.request;
+        submitRequest(std::move(request), sub.mode, {}, *lastDelivered + 1);
+
+        for (auto &message : messages) {
+            invokeSubscriptionCallbacks(sub, std::move(message));
+        }
+    }
+
+    void resumeSubscriptionAfterBatchError(const std::string &zmqTopic, mdp::Message &&response, std::optional<std::uint64_t> nextIndex = std::nullopt) {
+        const auto subIt = _subscriptions.find(zmqTopic);
+        if (subIt == _subscriptions.end()) {
+            return;
+        }
+        auto &sub      = subIt->second;
+        response.topic = sub.request.topic;
+        response.data.clear();
+        HTTP_DBG("Client: {}", response.error);
+
+        auto request = sub.request;
+        submitRequest(std::move(request), SubscriptionMode::Next, {}, nextIndex);
+        invokeSubscriptionCallbacks(sub, std::move(response));
+    }
+
+    void reportSubscriptionError(const std::string &zmqTopic, mdp::Message &&response) {
+        auto subIt = _subscriptions.find(zmqTopic);
+        if (subIt == _subscriptions.end()) {
+            return;
+        }
+        auto subscription = std::move(subIt->second);
+        _subscriptions.erase(subIt);
+        invokeSubscriptionCallbacks(subscription, std::move(response));
+    }
+
+    static void invokeSubscriptionCallbacks(Subscription &sub, mdp::Message &&response) {
         for (std::size_t i = 0; i < sub.callbacks.size(); ++i) {
+            if (!sub.callbacks[i]) {
+                continue;
+            }
             if (i < sub.callbacks.size() - 1) {
                 auto copy = response;
                 sub.callbacks[i](std::move(copy));
@@ -340,18 +492,48 @@ struct ClientSessionBase {
     }
 
     void startSubscription(client::Command &&command, SubscriptionMode mode = SubscriptionMode::Next) {
-        mdp::Topic topic;
+        const auto reportError = [](client::Command &failedCommand, std::string error) {
+            if (!failedCommand.callback) {
+                return;
+            }
+            mdp::Message response{};
+            response.command = mdp::Command::Notify;
+            response.topic   = failedCommand.topic;
+            response.error   = std::move(error);
+            failedCommand.callback(response);
+        };
+
+        std::optional<std::uint64_t> initialIndex;
+        std::string                  key;
+        bool                         usesBatch = false;
         try {
-            topic = mdp::Topic::fromMdpTopic(command.topic);
+            const auto parsedIndex = takeInitialLongPollingIndex(command, mode);
+            if (!parsedIndex.has_value()) {
+                reportError(command, parsedIndex.error());
+                return;
+            }
+            initialIndex = *parsedIndex;
+            usesBatch    = usesLongPollingBatch(command.topic);
+            key          = subscriptionKey(command.topic);
         } catch (const std::exception &e) {
             HTTP_DBG("Client::startSubscription: Could not parse topic '{}': {}", command.topic.str(), e.what());
+            reportError(command, e.what());
+            return;
+        }
+
+        const auto [subIt, inserted] = _subscriptions.try_emplace(key, Subscription{});
+        if (!inserted && (usesBatch || usesLongPollingBatch(subIt->second.request.topic))) {
+            reportError(command, "A subscription for this topic is already active; concurrent batch subscriptions are not supported");
             return;
         }
-        const auto [subIt, inserted] = _subscriptions.try_emplace(topic.toZmqTopic(), Subscription{});
-        subIt->second.request        = command;
+        subIt->second.request = command;
+        subIt->second.mode    = mode;
         subIt->second.callbacks.emplace_back(command.callback);
         if (inserted) {
-            submitRequest(std::move(command), mode, {}, {});
+            if (initialIndex.has_value() && *initialIndex > 0) {
+                subIt->second.lastReceivedLongPollingIdx = *initialIndex - 1;
+            }
+            submitRequest(std::move(command), mode, {}, initialIndex);
         }
     }
 
@@ -359,24 +541,25 @@ struct ClientSessionBase {
         // TODO a single unsubscribe cancels this also in case of multiple subscriptions when the client is shared
         // inside an application. Would be great if we could selectively unsubscribe certain callbacks and finally
         // stop the subscription when all callbacks are removed.
-        mdp::Topic topic;
+        std::string key;
         try {
-            topic = mdp::Topic::fromMdpTopic(command.topic);
+            key = subscriptionKey(command.topic);
         } catch (const std::exception &e) {
             HTTP_DBG("Client::stopSubscription: Could not parse topic '{}': {}", command.topic.str(), e.what());
             return;
         };
-        if (auto subIt = _subscriptions.find(topic.toZmqTopic()); subIt != _subscriptions.end()) {
+        if (auto subIt = _subscriptions.find(key); subIt != _subscriptions.end()) {
             // Cancel all requests for this topic
             auto reqIt = _requestsByStreamId.begin();
             while (reqIt != _requestsByStreamId.end()) {
-                if (reqIt->second.request.topic == command.topic) {
+                if (reqIt->second.request.command == mdp::Command::Subscribe && reqIt->second.normalizedTopic == key) {
                     self().cancelStream(reqIt->first);
                     reqIt = _requestsByStreamId.erase(reqIt);
                 } else {
                     ++reqIt;
                 }
             }
+            _subscriptions.erase(subIt);
         }
     }
 };
@@ -627,7 +810,8 @@ struct RestClient : public ClientBase {
         } {
         _worker = std::jthread([queue = _requestQueue, sslSettings = _sslSettings, mimeType = _mimeType](std::stop_token stopToken) {
             auto preferredMimeType = [&mimeType](const URI<> &topic) {
-                if (const auto contentTypeHeader = topic.queryParamMap().find("contentType"); contentTypeHeader != topic.queryParamMap().end() && contentTypeHeader->second) {
+                const auto &parameters = topic.queryParamMap();
+                if (const auto contentTypeHeader = parameters.find("contentType"); contentTypeHeader != parameters.end() && contentTypeHeader->second) {
                     return contentTypeHeader->second.value();
                 }
                 return std::string{ mimeType.typeName() };
@@ -641,7 +825,7 @@ struct RestClient : public ClientBase {
                 if (!cmd.callback) {
                     return;
                 }
-                mdp::Message msg;
+                mdp::Message msg{};
                 msg.protocolName    = cmd.topic.scheme().value_or("");
                 msg.arrivalTime     = std::chrono::system_clock::now();
                 msg.command         = mdp::Command::Final;
@@ -659,12 +843,18 @@ struct RestClient : public ClientBase {
                     switch (cmd.command) {
                     case mdp::Command::Get:
                     case mdp::Command::Set: {
+                        std::string preferred;
+                        try {
+                            preferred = preferredMimeType(cmd.topic);
+                        } catch (const std::exception &e) {
+                            reportError(cmd, e.what());
+                            continue;
+                        }
                         auto session = ensureSession(ssl_ctx, sessions, sslSettings, cmd.topic);
                         if (!session) {
                             reportError(cmd, std::format("Could not create REST session for endpoint '{}': {}", cmd.topic.str(), session.error()));
                             continue;
                         }
-                        auto preferred = preferredMimeType(cmd.topic);
                         session.value()->submitRequest(std::move(cmd), mode, std::move(preferred), {});
                     } break;
                     case mdp::Command::Subscribe: {
diff --git a/src/client/test/emscripten_rest_client/EmscriptenRestClientTest.cpp b/src/client/test/emscripten_rest_client/EmscriptenRestClientTest.cpp
index aaeb0696..095b79d0 100644
--- a/src/client/test/emscripten_rest_client/EmscriptenRestClientTest.cpp
+++ b/src/client/test/emscripten_rest_client/EmscriptenRestClientTest.cpp
@@ -12,6 +12,7 @@
 #include <source_location>
 #include <string>
 #include <string_view>
+#include <utility>
 
 #include <RestClient.hpp>
 
@@ -21,18 +22,27 @@ using namespace opencmw::client;
 namespace {
 
 constexpr int kStreamACount = 5;
+constexpr int kStreamBCount = 1;
+constexpr int kStreamCCount = 5;
+constexpr int kStreamAFirst = 7;
+constexpr int kStreamBFirst = 5;
+constexpr int kStreamCFirst = 5;
+constexpr int kStreamCStart = 3;
 
 struct TestState {
-    int                                   failures{};
+    std::atomic_int                       failures{};
     int                                   initialRunningWorkers{};
     int                                   initialUnusedWorkers{};
     std::optional<RestClient>             client;
     std::atomic_int                       messagesA{};
     std::atomic_int                       messagesB{};
+    std::atomic_int                       messagesC{};
+    std::atomic_int                       errorsC{};
     std::atomic_bool                      sawMainThread{};
     std::atomic_bool                      sawWorkerThread{};
     std::string                           receivedA;
     std::string                           receivedB;
+    std::string                           receivedC;
     int                                   callbackCountAtCleanup{};
     std::chrono::steady_clock::time_point deadline;
 };
@@ -48,6 +58,39 @@ void check(TestState &state, bool condition, std::string_view failure, std::sour
     }
 }
 
+void checkBatchSubscriptionOverlap(TestState &state) {
+    for (const auto &[firstQuery, secondQuery] : {
+                 std::pair{ "?LongPollingBatch=2", "?LongPollingIdx=5&LongPollingBatch=3" },
+                 std::pair{ "?LongPollingBatch=2", "?LongPollingIdx=5" },
+                 std::pair{ "", "?LongPollingIdx=5&LongPollingBatch=3" } }) {
+        client::detail::RestWorkerState worker{ MIME::BINARY };
+        const URI<STRICT>              firstTopic(std::format("http://localhost/stream{}", firstQuery));
+        Command                        first;
+        first.command = mdp::Command::Subscribe;
+        first.topic   = firstTopic;
+        worker._subscriptions.emplace(1, client::detail::SubscriptionState{ .command = std::move(first) });
+        worker._nextSubscriptionId = 2;
+
+        int     errors = 0;
+        Command second;
+        second.command  = mdp::Command::Subscribe;
+        second.topic    = URI<STRICT>(std::format("http://localhost/stream{}", secondQuery));
+        second.callback = [&](const mdp::Message &message) {
+            check(state, !message.error.empty(), "overlapping subscription did not report an error");
+            ++errors;
+        };
+        worker.startSubscription(std::move(second));
+
+        check(state, errors == 1, "overlapping subscription was not rejected exactly once");
+        check(state, worker._subscriptions.size() == 1, "overlapping subscription was added");
+        check(state, worker._subscriptions.at(1).command.topic == firstTopic, "overlapping subscription changed the active topic");
+        check(state, worker._activeFetches.empty(), "overlapping subscription started a request");
+        while (!worker._activeFetches.empty()) {
+            worker.closeFetch(worker._activeFetches.begin()->first);
+        }
+    }
+}
+
 // PThread worker counts are Emscripten internals used only for this leak check.
 int runningWorkerCount() noexcept {
     return EM_ASM_INT({ return PThread.runningWorkers.length; });
@@ -69,6 +112,14 @@ std::string testPayload(int index) {
     return expected;
 }
 
+std::string expectedPayloads(int first, int count) {
+    std::string expected;
+    for (int index = first; index < first + count; ++index) {
+        expected += testPayload(index);
+    }
+    return expected;
+}
+
 void recordCallbackThread(TestState &state) {
     if (emscripten_is_main_runtime_thread()) {
         state.sawMainThread.store(true, std::memory_order_relaxed);
@@ -78,33 +129,31 @@ void recordCallbackThread(TestState &state) {
 }
 
 void reportAndExit(const TestState &state) {
-    std::println("=== {} ({} failure{}) ===", state.failures == 0 ? "PASSED" : "FAILED", state.failures, state.failures == 1 ? "" : "s");
-    emscripten_force_exit(state.failures == 0 ? 0 : 1);
+    const int failures = state.failures.load();
+    std::println("=== {} ({} failure{}) ===", failures == 0 ? "PASSED" : "FAILED", failures, failures == 1 ? "" : "s");
+    emscripten_force_exit(failures == 0 ? 0 : 1);
 }
 
 void finishTest(void *data) {
-    constexpr int kStreamAFirst = 7;
-    constexpr int kStreamBFirst = 5;
-
-    auto         &state         = testState(data);
-    const int     messagesA     = state.messagesA.load(std::memory_order_acquire);
-    const int     messagesB     = state.messagesB.load(std::memory_order_acquire);
-    const int     callbacks     = messagesA + messagesB;
+    auto     &state     = testState(data);
+    const int messagesA = state.messagesA.load(std::memory_order_acquire);
+    const int messagesB = state.messagesB.load(std::memory_order_acquire);
+    const int messagesC = state.messagesC.load(std::memory_order_acquire);
+    const int errorsC   = state.errorsC.load(std::memory_order_acquire);
+    const int callbacks = messagesA + messagesB + messagesC + errorsC;
 
     check(state, state.callbackCountAtCleanup == callbacks, std::format("callback count changed after cleanup ({} to {})", state.callbackCountAtCleanup, callbacks));
     check(state, workerPoolRestored(state), "worker count changed after cleanup");
     check(state, messagesA == kStreamACount, std::format("stream A delivered {} messages, expected {}", messagesA, kStreamACount));
-    check(state, messagesB == 1, std::format("stream B delivered {} messages, expected 1", messagesB));
+    check(state, messagesB == kStreamBCount, std::format("stream B delivered {} messages, expected {}", messagesB, kStreamBCount));
+    check(state, messagesC == kStreamCCount, std::format("stream C delivered {} messages, expected {}", messagesC, kStreamCCount));
+    check(state, errorsC == 1, std::format("stream C reported {} errors, expected one malformed batch", errorsC));
     check(state, !state.sawMainThread.load(std::memory_order_relaxed), "a callback ran on the browser main thread");
     check(state, state.sawWorkerThread.load(std::memory_order_relaxed), "no callback ran on the REST worker");
 
-    std::string expectedA;
-    for (int index = kStreamAFirst; index < kStreamAFirst + kStreamACount; ++index) {
-        expectedA += testPayload(index);
-    }
-    const std::string expectedB = testPayload(kStreamBFirst);
-    check(state, state.receivedA == expectedA, std::format("stream A payload differs ({} bytes, expected {})", state.receivedA.size(), expectedA.size()));
-    check(state, state.receivedB == expectedB, std::format("stream B payload differs ({} bytes, expected {})", state.receivedB.size(), expectedB.size()));
+    check(state, state.receivedA == expectedPayloads(kStreamAFirst, kStreamACount), "stream A payload differs");
+    check(state, state.receivedB == expectedPayloads(kStreamBFirst, kStreamBCount), "stream B payload differs");
+    check(state, state.receivedC == expectedPayloads(kStreamCFirst, kStreamCCount), "stream C payload differs");
 
     reportAndExit(state);
 }
@@ -117,9 +166,11 @@ void waitForDelivery(void *data) {
 
     auto          &state              = testState(data);
     const auto     now                = std::chrono::steady_clock::now();
-    if (state.messagesA.load(std::memory_order_acquire) >= kStreamACount && state.messagesB.load(std::memory_order_acquire) >= 1) {
+    if (state.messagesA.load(std::memory_order_acquire) >= kStreamACount
+            && state.messagesB.load(std::memory_order_acquire) >= kStreamBCount
+            && state.messagesC.load(std::memory_order_acquire) >= kStreamCCount) {
         emscripten_cancel_main_loop();
-        // Allow the delayed sixth response to expose a failed unsubscribe.
+        // Allow delayed A/C responses to expose failed unsubscribes; B keeps its poll open.
         emscripten_set_timeout([](void *callbackData) {
             constexpr auto kCleanupTimeout = std::chrono::seconds{ 10 };
 
@@ -150,7 +201,7 @@ void waitForCleanup(void *data) {
     auto          &state            = testState(data);
     const auto     now              = std::chrono::steady_clock::now();
     if (workerPoolRestored(state)) {
-        state.callbackCountAtCleanup = state.messagesA.load(std::memory_order_acquire) + state.messagesB.load(std::memory_order_acquire);
+        state.callbackCountAtCleanup = state.messagesA.load(std::memory_order_acquire) + state.messagesB.load(std::memory_order_acquire) + state.messagesC.load(std::memory_order_acquire) + state.errorsC.load(std::memory_order_acquire);
         emscripten_set_timeout(&finishTest, kStabilityWindow.count(), data);
         return;
     }
@@ -181,11 +232,14 @@ int main(int argc, char **argv) {
         return 2;
     }
 
-    const URI<STRICT> topicA(std::format("http://127.0.0.1:{}/streamA", port));
+    const URI<STRICT> topicA(std::format("http://127.0.0.1:{}/streamA?b=2&LongPollingIdx=Next&value=%41&a=1", port));
     const URI<STRICT> topicB(std::format("http://127.0.0.1:{}/streamB", port));
+    const URI<STRICT> topicC(std::format("http://127.0.0.1:{}/streamC?LongPollingIdx={}&LongPollingBatch=AllAvailable", port, kStreamCStart));
 
     std::println("Emscripten RestClient integration test (server on port {})", port);
 
+    checkBatchSubscriptionOverlap(state);
+
     state.initialRunningWorkers = runningWorkerCount();
     state.initialUnusedWorkers  = unusedWorkerCount();
     state.deadline              = std::chrono::steady_clock::now() + kDeliveryTimeout;
@@ -201,7 +255,7 @@ int main(int argc, char **argv) {
         if (test->messagesA.load(std::memory_order_relaxed) == kStreamACount - 1) {
             Command unsubscribe;
             unsubscribe.command = mdp::Command::Unsubscribe;
-            unsubscribe.topic   = topicA;
+            unsubscribe.topic   = URI<STRICT>(std::format("http://127.0.0.1:{}/streamA?value=%41&a=1&b=2", topicA.port().value()));
             test->client->request(std::move(unsubscribe));
         }
         test->messagesA.fetch_add(1, std::memory_order_release);
@@ -218,5 +272,34 @@ int main(int argc, char **argv) {
     };
     state.client->request(std::move(subscribeB));
 
+    Command subscribeC;
+    subscribeC.command  = mdp::Command::Subscribe;
+    subscribeC.topic    = topicC;
+    subscribeC.callback = [test = &state, port](const mdp::Message &message) {
+        recordCallbackThread(*test);
+        if (!message.error.empty() && message.data.empty()) {
+            check(*test, test->messagesC.load(std::memory_order_relaxed) == 2, "stream C error did not follow its first batch");
+            test->errorsC.fetch_add(1, std::memory_order_release);
+            return;
+        }
+        const auto index = kStreamCFirst + test->messagesC.load(std::memory_order_relaxed);
+        if (index == kStreamCFirst) {
+            check(*test, !message.error.empty(), "stream C did not report its initial index gap");
+        } else {
+            check(*test, message.error.empty(), "stream C reported an unexpected warning");
+        }
+        check(*test, message.topic == URI<STRICT>(std::format("/streamC?sample={}", index)), "stream C received the wrong topic");
+        check(*test, message.serviceName == std::format("/streamC-service-{}", index), "stream C received the wrong service name");
+        test->receivedC += message.data.asString();
+        if (test->messagesC.load(std::memory_order_relaxed) == kStreamCCount - 1) {
+            Command unsubscribe;
+            unsubscribe.command = mdp::Command::Unsubscribe;
+            unsubscribe.topic   = URI<STRICT>(std::format("http://127.0.0.1:{}/streamC", port));
+            test->client->request(std::move(unsubscribe));
+        }
+        test->messagesC.fetch_add(1, std::memory_order_release);
+    };
+    state.client->request(std::move(subscribeC));
+
     emscripten_set_main_loop_arg(&waitForDelivery, &state, 0, EM_TRUE);
 }
diff --git a/src/client/test/emscripten_rest_client/run.py b/src/client/test/emscripten_rest_client/run.py
index c6423edd..4b9c8768 100644
--- a/src/client/test/emscripten_rest_client/run.py
+++ b/src/client/test/emscripten_rest_client/run.py
@@ -6,20 +6,36 @@
 import threading
 from urllib.parse import parse_qs, urlparse
 
-# The delayed sixth stream-A response makes a failed unsubscribe observable.
 STREAMS = {
     "/streamA": (7, 8, 9, 10, 11, 12),
-    "/streamB": (5,),
+    "/streamB": (5,),  # Leave the following poll open until client.stop().
+    "/streamC": (5, 6, 7, 8, 9, 10),
 }
-PROBE_INDEX = ("/streamA", 12)
+PROBE_INDICES = {("/streamA", 12), ("/streamC", 10)}
+STREAM_C_BATCHES = {5: (5, 6), 7: (7, 8, 9), 10: (10,)}
 
 def payload(index):
     return "{}:{}".format(index, "".join(str(i) for i in range(100))).encode()
 
+def batch_payload(path, indices, boundary):
+    body = bytearray()
+    for index in indices:
+        data = payload(index)
+        body.extend("--{}\r\n".format(boundary).encode())
+        body.extend("x-opencmw-long-polling-idx: {}\r\n".format(index).encode())
+        body.extend("x-opencmw-topic: {}?sample={}\r\n".format(path, index).encode())
+        body.extend("x-opencmw-service-name: {}-service-{}\r\n".format(path, index).encode())
+        body.extend("content-length: {}\r\n\r\n".format(len(data)).encode())
+        body.extend(data)
+        body.extend(b"\r\n")
+    body.extend("--{}--\r\n".format(boundary).encode())
+    return bytes(body)
+
 HOLD_SECONDS = 30.0
 PROBE_DELAY_SECONDS = 0.5
 
 stopping = threading.Event()
+stream_c_recovery = threading.Event()
 
 class Handler(http.server.BaseHTTPRequestHandler):
     protocol_version = "HTTP/1.1"
@@ -42,28 +58,54 @@ def do_GET(self):
             self._respond(404, b"unknown stream")
             return
 
-        index = parse_qs(parsed.query).get("LongPollingIdx", [""])[0]
+        params = parse_qs(parsed.query)
+        index = params.get("LongPollingIdx", [""])[0]
+        batch = params.get("LongPollingBatch", [None])[0]
         if index == "Next":
-            self._redirect(parsed.path, min(indices))
+            if parsed.path == "/streamC":
+                # Recovery must request Next; starting with Next would miss messages 5 and 6.
+                stream_c_recovery.set()
+                self._redirect(parsed.path, 7, batch)
+            else:
+                self._redirect(parsed.path, min(indices), batch)
             return
         if not index.isdigit():
             self._respond(400, b"malformed LongPollingIdx")
             return
 
-        if int(index) not in indices:
+        numeric_index = int(index)
+        if parsed.path == "/streamC" and batch == "AllAvailable":
+            # Request 3 starts at the oldest buffered message, 5.
+            numeric_index = max(numeric_index, min(indices))
+        if numeric_index not in indices:
             stopping.wait(HOLD_SECONDS)
             self._respond(504, b"")
             return
-        if (parsed.path, int(index)) == PROBE_INDEX:
+        if (parsed.path, numeric_index) in PROBE_INDICES:
             stopping.wait(PROBE_DELAY_SECONDS)
-        self._respond(200, payload(int(index)))
+
+        if batch is not None:
+            selected = STREAM_C_BATCHES.get(numeric_index) if parsed.path == "/streamC" and batch == "AllAvailable" else None
+            if selected is None:
+                self._respond(400, b"unexpected batch request")
+                return
+            if parsed.path == "/streamC" and numeric_index == 7 and not stream_c_recovery.is_set():
+                self._respond(200, b"not a multipart response", "multipart/mixed")
+                return
+            boundary = "opencmw-long-polling-multipart-boundary-{}".format(numeric_index)
+            self._respond(200, batch_payload(parsed.path, selected, boundary), "multipart/mixed; boundary={}".format(boundary))
+            return
+
+        self._respond(200, payload(numeric_index))
 
     def _common_headers(self):
         self.send_header("Access-Control-Allow-Origin", "*")
 
-    def _redirect(self, path, index):
+    def _redirect(self, path, index, batch):
         # Absolute, because xhr2 does not resolve a relative Location against the request URL.
         location = "http://{}{}?LongPollingIdx={}".format(self.headers["Host"], path, index)
+        if batch is not None:
+            location += "&LongPollingBatch={}".format(batch)
         try:
             self.send_response(302)
             self._common_headers()
@@ -73,11 +115,11 @@ def _redirect(self, path, index):
         except (BrokenPipeError, ConnectionResetError):
             pass
 
-    def _respond(self, code, body):
+    def _respond(self, code, body, content_type="application/octet-stream"):
         try:
             self.send_response(code)
             self._common_headers()
-            self.send_header("Content-Type", "application/octet-stream")
+            self.send_header("Content-Type", content_type)
             self.send_header("Content-Length", str(len(body)))
             self.end_headers()
             self.wfile.write(body)
diff --git a/src/client/test/nghttp2_tests.cpp b/src/client/test/nghttp2_tests.cpp
index e40bb4b7..bc34658c 100644
--- a/src/client/test/nghttp2_tests.cpp
+++ b/src/client/test/nghttp2_tests.cpp
@@ -110,6 +110,25 @@ static std::string normalize(URI<> uri) {
     return opencmw::mdp::Topic::fromMdpTopic(uri).toZmqTopic();
 }
 
+struct TestNativeClientSession : client::detail::ClientSessionBase<TestNativeClientSession, int> {
+    int                      nextStreamId = 1;
+    std::vector<int>         cancelledStreams;
+    std::vector<std::string> requestPaths;
+
+    int                      submitRequestImpl(const std::vector<nghttp2_nv> &headers, const IoBuffer *) {
+        for (const auto &header : headers) {
+            if (std::string_view(reinterpret_cast<const char *>(header.name), header.namelen) == ":path") {
+                requestPaths.emplace_back(reinterpret_cast<const char *>(header.value), header.valuelen);
+            }
+        }
+        return nextStreamId++;
+    }
+
+    void cancelStream(int streamId) {
+        cancelledStreams.push_back(streamId);
+    }
+};
+
 TEST_CASE("Basic Client Constructor and API Tests", "[http2]") {
     using namespace opencmw::client;
 
@@ -126,6 +145,110 @@ TEST_CASE("Basic Client Constructor and API Tests", "[http2]") {
     REQUIRE(client3.verifySslPeers() == false);
 }
 
+TEST_CASE("Native long-poll subscriptions", "[http2][long-polling]") {
+    using namespace opencmw::client;
+
+    SECTION("Continue after an invalid batch") {
+        TestNativeClientSession   session;
+        std::vector<mdp::Message> responses;
+
+        Command                   subscription;
+        subscription.command  = mdp::Command::Subscribe;
+        subscription.topic    = URI<>("http://localhost/batch?LongPollingIdx=0&LongPollingBatch=2");
+        subscription.callback = [&](const mdp::Message &message) { responses.push_back(message); };
+        session.startSubscription(std::move(subscription));
+
+        REQUIRE(session._subscriptions.size() == 1);
+        REQUIRE(session._requestsByStreamId.size() == 1);
+        const auto streamId    = session._requestsByStreamId.begin()->first;
+        auto      &request     = session._requestsByStreamId.begin()->second;
+        request.responseStatus = "200";
+        request.longPollingIdx = 0;
+        request.payload        = "not a multipart response";
+
+        CHECK(session.processResponse(streamId) == 0);
+        REQUIRE(responses.size() == 1);
+        CHECK_FALSE(responses.front().error.empty());
+        CHECK(responses.front().data.empty());
+        REQUIRE(session._subscriptions.size() == 1);
+        REQUIRE(session._requestsByStreamId.size() == 1);
+        REQUIRE(session.requestPaths.size() == 2);
+        const URI<> retryPath(session.requestPaths.back());
+        CHECK(retryPath.queryParamMap().at("LongPollingIdx") == "Next");
+        CHECK(retryPath.queryParamMap().at("LongPollingBatch") == "2");
+    }
+
+    SECTION("Single-message polling after indexed HTTP 500") {
+        TestNativeClientSession   session;
+        std::vector<mdp::Message> responses;
+
+        Command                   subscription;
+        subscription.command  = mdp::Command::Subscribe;
+        subscription.topic    = URI<>("http://localhost/single?LongPollingIdx=0");
+        subscription.callback = [&](const mdp::Message &message) { responses.push_back(message); };
+        session.startSubscription(std::move(subscription));
+
+        REQUIRE(session._requestsByStreamId.size() == 1);
+        const auto streamId    = session._requestsByStreamId.begin()->first;
+        auto      &request     = session._requestsByStreamId.begin()->second;
+        request.responseStatus = "500";
+        request.longPollingIdx = 0;
+        request.payload        = "server error";
+
+        CHECK(session.processResponse(streamId) == 0);
+        REQUIRE(responses.size() == 1);
+        CHECK_FALSE(responses.front().error.empty());
+        CHECK(session._subscriptions.size() == 1);
+        REQUIRE(session._requestsByStreamId.size() == 1);
+        REQUIRE(session.requestPaths.size() == 2);
+        const URI<> nextPath(session.requestPaths.back());
+        CHECK(nextPath.queryParamMap().at("LongPollingIdx") == "1");
+    }
+
+    SECTION("Reject conflicting subscriptions") {
+        TestNativeClientSession   session;
+        std::vector<mdp::Message> responses;
+
+        Command                   first;
+        first.command  = mdp::Command::Subscribe;
+        first.topic    = URI<>("http://localhost/batch?LongPollingIdx=0&LongPollingBatch=2");
+        first.callback = [](const mdp::Message &) {};
+        session.startSubscription(std::move(first));
+
+        Command second;
+        second.command  = mdp::Command::Subscribe;
+        second.topic    = URI<>("http://localhost/batch?LongPollingIdx=5&LongPollingBatch=3");
+        second.callback = [&](const mdp::Message &message) { responses.push_back(message); };
+        session.startSubscription(std::move(second));
+
+        REQUIRE(session._subscriptions.size() == 1);
+        REQUIRE(session._requestsByStreamId.size() == 1);
+        const auto &parameters = session._subscriptions.begin()->second.request.topic.queryParamMap();
+        REQUIRE(parameters.at(std::string(long_polling::kBatchParameter)));
+        CHECK(*parameters.at(std::string(long_polling::kBatchParameter)) == "2");
+        REQUIRE(responses.size() == 1);
+        CHECK_FALSE(responses.front().error.empty());
+        CHECK(responses.front().id == 0);
+    }
+
+    SECTION("Reject invalid subscription queries") {
+        TestNativeClientSession   session;
+        std::vector<mdp::Message> responses;
+
+        Command                   subscription;
+        subscription.command  = mdp::Command::Subscribe;
+        subscription.topic    = URI<>("http://localhost/batch?range=1+2&LongPollingIdx=0&LongPollingBatch=2");
+        subscription.callback = [&](const mdp::Message &message) { responses.push_back(message); };
+
+        CHECK_NOTHROW(session.startSubscription(std::move(subscription)));
+        CHECK(session._subscriptions.empty());
+        CHECK(session._requestsByStreamId.empty());
+        REQUIRE(responses.size() == 1);
+        CHECK_FALSE(responses.front().error.empty());
+        CHECK(responses.front().id == 0);
+    }
+}
+
 TEST_CASE("GET HTTP", "[http2]") {
     using namespace opencmw::client;
 
@@ -154,6 +277,9 @@ TEST_CASE("GET HTTP", "[http2]") {
         ensureMessageReceived(server, stopToken, messages);
     });
 
+    std::vector<mdp::Message> rejectedResponses;
+    std::atomic<int>         rejectedCount = 0;
+
     // Client using plain http
     RestClient  http;
     Stopper     stopper(serverThread.get_stop_source());
@@ -161,6 +287,32 @@ TEST_CASE("GET HTTP", "[http2]") {
 
     std::atomic<int> responseCount = 0;
 
+    // Rejected queries must report errors without stopping the worker or reaching the server.
+    const URI<> rejectedTopic(std::format("http://localhost:{}/sayhello?ctx=a,b", kServerPort));
+    for (const auto command : { mdp::Command::Get, mdp::Command::Set }) {
+        client::Command invalid{};
+        invalid.command         = command;
+        invalid.topic           = rejectedTopic;
+        invalid.clientRequestID = IoBuffer(command == mdp::Command::Get ? "invalid-get" : "invalid-set");
+        invalid.callback        = [&rejectedResponses, &rejectedCount](const mdp::Message &response) {
+            rejectedResponses.push_back(response);
+            ++rejectedCount;
+        };
+        http.request(std::move(invalid));
+    }
+    REQUIRE(waitFor(rejectedCount, 2));
+    REQUIRE(rejectedResponses.size() == 2);
+    for (std::size_t i = 0; i < rejectedResponses.size(); ++i) {
+        const auto &response = rejectedResponses[i];
+        CHECK(response.id == 0);
+        CHECK(response.command == mdp::Command::Final);
+        CHECK(response.topic == rejectedTopic);
+        CHECK(response.clientRequestID.asString() == (i == 0 ? "invalid-get" : "invalid-set"));
+        CHECK(response.error.contains("URI query contains illegal characters"));
+        CHECK(response.data.empty());
+    }
+
+    // A valid request on the same client verifies that its worker continues after both errors.
     client::Command  req0;
     req0.command         = mdp::Command::Get;
     req0.clientRequestID = opencmw::IoBuffer("0");
@@ -195,6 +347,25 @@ TEST_CASE("GET HTTP", "[http2]") {
     REQUIRE(waitFor(responseCount, 2));
 }
 
+TEST_CASE("Invalid redirect queries are reported without throwing", "[http2]") {
+    client::detail::Http2ClientSession session{ opencmw::rest::detail::TcpSocket{} };
+    std::vector<mdp::Message>          responses;
+
+    auto &request            = session._requestsByStreamId[1];
+    request.request.command  = mdp::Command::Get;
+    request.request.topic    = URI<>("http://localhost/service");
+    request.request.callback = [&](const mdp::Message &message) { responses.push_back(message); };
+    REQUIRE(session.addHeader(1, ":status", "302"));
+    REQUIRE(session.addHeader(1, "location", "/service?LongPollingIdx=1&ctx=a,b"));
+
+    int result = 0;
+    REQUIRE_NOTHROW(result = session.processResponse(1));
+    CHECK(result == NGHTTP2_ERR_TEMPORAL_CALLBACK_FAILURE);
+    REQUIRE(responses.size() == 1);
+    CHECK_FALSE(responses.front().error.empty());
+    CHECK(session._requestsByStreamId.empty());
+}
+
 TEST_CASE("HTTPS", "[http2]") {
     using namespace opencmw::client;
 
@@ -476,16 +647,18 @@ TEST_CASE("REST client survives hostname resolution failure", "[http2]") {
 }
 
 TEST_CASE("Long polling example", "[http2]") {
-    constexpr int kFooMessages = 50;
+    const bool useBatch = GENERATE(false, true);
+    CAPTURE(useBatch);
+    constexpr int kFooMessages = 6;
 
     auto          brokerThread = std::jthread([](std::stop_token stopToken) {
-        RestServer server;
+        RestServer                server;
         majordomo::rest::Settings settings{ .port = kServerPort, .protocols = majordomo::rest::Http2 };
         REQUIRE(server.bind(settings));
 
-        const auto                  topic = URI<>("/foo?param1=1&param2=foo%2Fbar");
+        const auto          topic = URI<>("/foo?param1=1&param2=foo%2Fbar");
 
-        std::deque<Message>         messages;
+        std::deque<Message> messages;
         ensureMessageReceived(server, stopToken, messages);
         REQUIRE(messages.size() >= 1);
         const auto req0 = std::move(messages[0]);
@@ -498,8 +671,8 @@ TEST_CASE("Long polling example", "[http2]") {
         for (int i = 0; i < kFooMessages; ++i) {
             Message notify;
             notify.command     = mdp::Command::Notify;
-            notify.serviceName = "/foo";
-            notify.topic       = topic;
+            notify.serviceName = std::format("/foo-service-{}", i);
+            notify.topic       = URI<>(std::format("/foo?sample={}", i));
             auto data          = std::to_string(i);
             notify.data        = opencmw::IoBuffer(data.data(), data.size());
             server.handleNotification(opencmw::mdp::Topic::fromMdpTopic(topic), std::move(notify));
@@ -541,15 +714,16 @@ TEST_CASE("Long polling example", "[http2]") {
 
         client::RestClient       client;
         opencmw::client::Command sub;
-        sub.command  = mdp::Command::Subscribe;
+        sub.command         = mdp::Command::Subscribe;
         sub.clientRequestID = opencmw::IoBuffer("0");
-        sub.topic           = URI<>(std::format("http://localhost:{}/foo?param1=1&param2=foo%2fbar", kServerPort));
-        sub.callback = [&responseCount](const mdp::Message &msg) {
+        sub.topic           = URI<>(std::format("http://localhost:{}/foo?param1=1&param2=foo%2fbar{}", kServerPort, useBatch ? "&LongPollingIdx=0&LongPollingBatch=3" : ""));
+        sub.callback        = [&responseCount](const mdp::Message &msg) {
             REQUIRE(msg.command == mdp::Command::Notify);
             REQUIRE(msg.error == "");
             REQUIRE(msg.data.asString() == std::to_string(responseCount));
             REQUIRE(msg.protocolName == "http");
-            REQUIRE(normalize(msg.topic) == normalize(URI<>("/foo?param1=1&param2=foo%2Fbar")));
+            REQUIRE(msg.topic == URI<>(std::format("/foo?sample={}", responseCount.load())));
+            REQUIRE(msg.serviceName == std::format("/foo-service-{}", responseCount.load()));
             responseCount++;
         };
         client.request(std::move(sub));
diff --git a/src/core/CMakeLists.txt b/src/core/CMakeLists.txt
index c30e9c26..f7d099b3 100644
--- a/src/core/CMakeLists.txt
+++ b/src/core/CMakeLists.txt
@@ -2,7 +2,7 @@
 add_library(core INTERFACE)
 target_include_directories(core INTERFACE $<BUILD_INTERFACE:${CMAKE_CURRENT_SOURCE_DIR}/include> $<INSTALL_INTERFACE:include/>)
 target_link_libraries(core INTERFACE $<BUILD_INTERFACE:mp-units::mp-units> refl-cpp::refl-cpp pthread)
-set_target_properties(core PROPERTIES PUBLIC_HEADER "include/URI.hpp;include/MIME.hpp")
+set_target_properties(core PROPERTIES PUBLIC_HEADER "include/URI.hpp;include/MIME.hpp;include/LongPollingBatch.hpp")
 
 install(
         TARGETS core
diff --git a/src/core/include/LongPollingBatch.hpp b/src/core/include/LongPollingBatch.hpp
new file mode 100644
index 00000000..30f2519e
--- /dev/null
+++ b/src/core/include/LongPollingBatch.hpp
@@ -0,0 +1,160 @@
+#ifndef OPENCMW_LONGPOLLINGBATCH_HPP
+#define OPENCMW_LONGPOLLINGBATCH_HPP
+
+#include <algorithm>
+#include <charconv>
+#include <cstdint>
+#include <expected>
+#include <limits>
+#include <optional>
+#include <string>
+#include <string_view>
+#include <vector>
+
+namespace opencmw::long_polling {
+
+inline constexpr std::string_view kIndexParameter = "LongPollingIdx";
+inline constexpr std::string_view kBatchParameter = "LongPollingBatch";
+inline constexpr std::string_view kAllAvailable   = "AllAvailable";
+// Payloads may contain a line matching this separator, so read each part by its content-length;
+// a generic MIME parser scanning for delimiter lines can mistake such a line for a part's end.
+inline constexpr std::string_view kBoundary       = "opencmw-long-polling-multipart-boundary";
+
+struct BatchPart {
+    std::uint64_t    index;
+    std::string_view topic;
+    std::string_view serviceName;
+    std::string_view payload;
+};
+
+inline std::expected<std::uint64_t, std::string> parseUnsigned(std::string_view value, std::string_view name) {
+    if (value.empty()) {
+        return std::unexpected(std::string("Malformed ") + std::string(name) + " ''");
+    }
+
+    std::uint64_t result{};
+    const auto [end, error] = std::from_chars(value.data(), value.data() + value.size(), result);
+    if (error != std::errc{} || end != value.data() + value.size()) {
+        return std::unexpected(std::string("Malformed ") + std::string(name) + " '" + std::string(value) + "'");
+    }
+    return result;
+}
+
+inline std::expected<std::string, std::string> encodeBatch(const std::vector<BatchPart> &parts) {
+    for (const auto &part : parts) {
+        if (part.topic.find_first_of("\r\n") != std::string_view::npos || part.serviceName.find_first_of("\r\n") != std::string_view::npos) {
+            return std::unexpected("Multipart topic and service name must not contain CR or LF");
+        }
+    }
+
+    std::string body;
+    for (const auto &[index, topic, serviceName, payload] : parts) {
+        body += "--";
+        body += kBoundary;
+        body += "\r\nx-opencmw-long-polling-idx: ";
+        body += std::to_string(index);
+        body += "\r\nx-opencmw-topic: ";
+        body += topic;
+        body += "\r\nx-opencmw-service-name: ";
+        body += serviceName;
+        body += "\r\ncontent-length: ";
+        body += std::to_string(payload.size());
+        body += "\r\n\r\n";
+        if (!payload.empty()) {
+            body.append(payload.data(), payload.size());
+        }
+        body += "\r\n";
+    }
+    body += "--";
+    body += kBoundary;
+    body += "--\r\n";
+    return body;
+}
+
+inline std::expected<std::vector<BatchPart>, std::string> decodeBatch(std::string_view body) {
+    const auto boundaryEnd = body.find("\r\n");
+    if (boundaryEnd == std::string_view::npos || !body.starts_with("--")) {
+        return std::unexpected("Missing multipart boundary");
+    }
+    const auto boundaryLine = body.substr(0, boundaryEnd);
+    if (boundaryLine.size() <= 2 || boundaryLine.size() > 72) {
+        return std::unexpected("Invalid multipart boundary length");
+    }
+    const std::string      boundary(boundaryLine);
+    const std::string      partBoundary  = boundary + "\r\n";
+    const std::string      finalBoundary = boundary + "--\r\n";
+    std::vector<BatchPart> parts;
+    std::size_t            position = 0;
+
+    while (position < body.size()) {
+        if (body.substr(position).starts_with(finalBoundary)) {
+            position += finalBoundary.size();
+            if (position != body.size()) {
+                return std::unexpected("Unexpected data after final multipart boundary");
+            }
+            return parts;
+        }
+        if (!body.substr(position).starts_with(partBoundary)) {
+            return std::unexpected("Missing multipart boundary");
+        }
+        position += partBoundary.size();
+
+        const auto headerEnd = body.find("\r\n\r\n", position);
+        if (headerEnd == std::string_view::npos) {
+            return std::unexpected("Incomplete multipart headers");
+        }
+
+        std::optional<std::uint64_t>    index;
+        std::optional<std::uint64_t>    contentLength;
+        std::optional<std::string_view> topic;
+        std::optional<std::string_view> serviceName;
+        auto                            headerPosition = position;
+        while (headerPosition < headerEnd) {
+            const auto lineEnd = body.find("\r\n", headerPosition);
+            const auto end     = std::min(lineEnd == std::string_view::npos ? headerEnd : lineEnd, headerEnd);
+            const auto line    = body.substr(headerPosition, end - headerPosition);
+            if (line.starts_with("x-opencmw-long-polling-idx: ")) {
+                auto parsed = parseUnsigned(line.substr(std::string_view("x-opencmw-long-polling-idx: ").size()), "multipart index");
+                if (!parsed.has_value()) {
+                    return std::unexpected(parsed.error());
+                }
+                index = *parsed;
+            } else if (line.starts_with("x-opencmw-topic: ")) {
+                topic = line.substr(std::string_view("x-opencmw-topic: ").size());
+            } else if (line.starts_with("x-opencmw-service-name: ")) {
+                serviceName = line.substr(std::string_view("x-opencmw-service-name: ").size());
+            } else if (line.starts_with("content-length: ")) {
+                auto parsed = parseUnsigned(line.substr(std::string_view("content-length: ").size()), "multipart content length");
+                if (!parsed.has_value()) {
+                    return std::unexpected(parsed.error());
+                }
+                contentLength = *parsed;
+            }
+            headerPosition = end + 2;
+        }
+        if (!index.has_value() || !topic.has_value() || !serviceName.has_value() || !contentLength.has_value()) {
+            return std::unexpected("Multipart part is missing its index, topic, service name, or content length");
+        }
+
+        position = headerEnd + 4;
+        if (*contentLength > body.size() - position) {
+            return std::unexpected("Multipart payload is shorter than its content length");
+        }
+        if (!parts.empty() && (parts.back().index == std::numeric_limits<std::uint64_t>::max() || *index != parts.back().index + 1)) {
+            return std::unexpected("Long-polling batch contains non-consecutive indices");
+        }
+        const auto payloadSize = static_cast<std::size_t>(*contentLength);
+        parts.push_back(BatchPart{ *index, *topic, *serviceName, body.substr(position, payloadSize) });
+        position += payloadSize;
+        if (!body.substr(position).starts_with("\r\n")) {
+            return std::unexpected("Multipart payload is not followed by a boundary");
+        }
+        position += 2;
+    }
+
+    return std::unexpected("Missing final multipart boundary");
+}
+
+} // namespace opencmw::long_polling
+
+#endif // OPENCMW_LONGPOLLINGBATCH_HPP
diff --git a/src/core/test/CMakeLists.txt b/src/core/test/CMakeLists.txt
index 0840bdc5..968d9638 100644
--- a/src/core/test/CMakeLists.txt
+++ b/src/core/test/CMakeLists.txt
@@ -7,6 +7,7 @@ set(core_test_sources
     collection_tests.cpp
     URI_tests.cpp
     MIME_tests.cpp
+    LongPollingBatch_tests.cpp
     ReaderWriterLock_tests.cpp
     SpinWait_tests.cpp
     TimingCtx_tests.cpp)
diff --git a/src/core/test/LongPollingBatch_tests.cpp b/src/core/test/LongPollingBatch_tests.cpp
new file mode 100644
index 00000000..f1dc11d1
--- /dev/null
+++ b/src/core/test/LongPollingBatch_tests.cpp
@@ -0,0 +1,95 @@
+#include <LongPollingBatch.hpp>
+
+#include <catch2/catch.hpp>
+
+#include <cstdint>
+#include <format>
+#include <string>
+#include <string_view>
+#include <vector>
+
+using namespace opencmw;
+
+TEST_CASE("long_polling::parseUnsigned", "[core][rest][long-polling]") {
+    for (const auto value : { std::string_view{}, std::string_view("-1"), std::string_view("1x"), std::string_view("18446744073709551616") }) {
+        CAPTURE(value);
+        CHECK_FALSE(long_polling::parseUnsigned(value, long_polling::kIndexParameter).has_value());
+    }
+    for (const std::uint64_t value : { 0, 5 }) {
+        const auto parsed = long_polling::parseUnsigned(std::to_string(value), long_polling::kIndexParameter);
+        REQUIRE(parsed.has_value());
+        CHECK(*parsed == value);
+    }
+}
+
+TEST_CASE("Batch encoding and decoding", "[core][rest][long-polling]") {
+    SECTION("Binary, empty and text payloads") {
+        const std::string                          payload = std::string(1, '\0') + std::format("\r\n--{}--\r\n", long_polling::kBoundary);
+        const std::vector<long_polling::BatchPart> original{
+            { 42, "/batch", "/batch", payload },
+            { 43, "/batch?sample=43", "/batch-service-43", "" },
+            { 44, "/batch?sample=44", "/batch-service-44", "next message" }
+        };
+        const auto encoded = long_polling::encodeBatch(original);
+        REQUIRE(encoded.has_value());
+
+        const auto decoded = long_polling::decodeBatch(*encoded);
+        REQUIRE(decoded.has_value());
+        REQUIRE(decoded->size() == original.size());
+        for (std::size_t i = 0; i < original.size(); ++i) {
+            CHECK((*decoded)[i].index == original[i].index);
+            CHECK((*decoded)[i].topic == original[i].topic);
+            CHECK((*decoded)[i].serviceName == original[i].serviceName);
+            CHECK((*decoded)[i].payload == original[i].payload);
+        }
+    }
+
+    SECTION("Boundary ending in --") {
+        constexpr std::string_view body = "--batch--\r\n"
+                                          "x-opencmw-long-polling-idx: 42\r\n"
+                                          "x-opencmw-topic: /batch\r\n"
+                                          "x-opencmw-service-name: /batch\r\n"
+                                          "content-length: 5\r\n\r\n"
+                                          "hello\r\n"
+                                          "--batch----\r\n";
+        const auto decoded = long_polling::decodeBatch(body);
+        REQUIRE(decoded.has_value());
+        REQUIRE(decoded->size() == 1);
+        CHECK(decoded->front().index == 42);
+        CHECK(decoded->front().payload == "hello");
+    }
+
+    SECTION("Empty batch, index gaps and duplicates") {
+        for (const auto &parts : std::vector<std::vector<long_polling::BatchPart>>{
+                     {},
+                     { { 4, "/batch", "/batch", "a" }, { 6, "/batch", "/batch", "b" } },
+                     { { 4, "/batch", "/batch", "a" }, { 4, "/batch", "/batch", "b" } } }) {
+            const auto encoded = long_polling::encodeBatch(parts);
+            REQUIRE(encoded.has_value());
+            CHECK_FALSE(long_polling::decodeBatch(*encoded).has_value());
+        }
+    }
+
+    SECTION("Incomplete multipart bodies") {
+        const auto encoded = long_polling::encodeBatch({ { 4, "/batch", "/batch", "data" } });
+        REQUIRE(encoded.has_value());
+        auto              missingTopic  = *encoded;
+        const std::string topicHeader   = "x-opencmw-topic: /batch\r\n";
+        const auto        topicPosition = missingTopic.find(topicHeader);
+        REQUIRE(topicPosition != std::string::npos);
+        missingTopic.erase(topicPosition, topicHeader.size());
+        for (const auto &body : {
+                     encoded->substr(0, encoded->find("\r\n\r\n") + 4 + 3),
+                     encoded->substr(0, encoded->rfind(std::format("--{}--\r\n", long_polling::kBoundary))),
+                     missingTopic }) {
+            CHECK_FALSE(long_polling::decodeBatch(body).has_value());
+        }
+    }
+
+    SECTION("Line breaks in topic and service name") {
+        for (const std::string_view invalid : { "/batch\rvalue", "/batch\nvalue", "/batch\r\nx-opencmw-topic: /other" }) {
+            CHECK_FALSE(long_polling::encodeBatch({ { 0, invalid, "/batch", "data" } }).has_value());
+            CHECK_FALSE(long_polling::encodeBatch({ { 0, "/batch", invalid, "data" } }).has_value());
+        }
+    }
+}
diff --git a/src/majordomo/include/majordomo/RestServer.hpp b/src/majordomo/include/majordomo/RestServer.hpp
index 65160aa8..17caa9c8 100644
--- a/src/majordomo/include/majordomo/RestServer.hpp
+++ b/src/majordomo/include/majordomo/RestServer.hpp
@@ -3,6 +3,7 @@
 
 #include "IoBuffer.hpp"
 #include "LoadTest.hpp"
+#include "LongPollingBatch.hpp"
 #include "MdpMessage.hpp"
 #include "MIME.hpp"
 #include "NgTcp2Util.hpp"
@@ -22,6 +23,7 @@
 #include <expected>
 #include <filesystem>
 #include <iterator>
+#include <limits>
 #include <memory>
 #include <netinet/in.h>
 #include <netinet/tcp.h>
@@ -115,7 +117,7 @@ inline std::expected<int, std::string> create_sock(Address &local_addr, std::str
     hints.ai_socktype = SOCK_DGRAM;
 
     addrinfo *res, *rp;
-    int       val = 1;
+    int       val   = 1;
 
     auto      paddr = addr == "*" ? nullptr : addr.data();
 
@@ -269,17 +271,18 @@ enum class RestMethod {
 inline RestMethod parseMethod(std::string_view methodString) {
     using enum RestMethod;
     return methodString == "OPTIONS" ? Options
-         : methodString == "PUT"  ? Post
-         : methodString == "POST" ? Post
-         : methodString == "GET"  ? Get
-                                  : Invalid;
+         : methodString == "PUT"     ? Post
+         : methodString == "POST"    ? Post
+         : methodString == "GET"     ? Get
+                                     : Invalid;
 }
 
 struct Request {
     std::vector<std::pair<std::string, std::string>> rawHeaders;
     mdp::Topic                                       topic;
     RestMethod                                       method = RestMethod::Invalid;
-    std::string                                      longPollIndex;
+    std::optional<std::string>                       longPollIndex;
+    std::optional<std::string>                       longPollBatch;
     std::string                                      contentType;
     std::string                                      accept;
     std::string                                      payload;
@@ -463,14 +466,31 @@ struct ResponseData {
     IoBuffer *bodyBuffer = nullptr;
 };
 
-constexpr int kHttpOk       = 200;
-constexpr int kHttpError    = 500;
-constexpr int kFileNotFound = 404;
+constexpr int kHttpOk         = 200;
+constexpr int kHttpBadRequest = 400;
+constexpr int kHttpError      = 500;
+constexpr int kFileNotFound   = 404;
+
+struct LongPollingBatchRequest {
+    enum class Mode {
+        None,
+        AllAvailable,
+        FixedCount
+    };
+
+    Mode        mode  = Mode::None;
+    std::size_t count = 1;
+};
 
 template<typename Derived, typename TStreamId>
 struct SessionBase {
-    using PendingRequest = std::tuple<std::uint64_t, TStreamId>;              // requestId, streamId
-    using PendingPoll    = std::tuple<std::string, std::uint64_t, TStreamId>; // zmqTopic, PollingIndex, streamId
+    using PendingRequest = std::tuple<std::uint64_t, TStreamId>; // requestId, streamId
+    struct PendingPoll {
+        std::string             zmqTopic;
+        std::uint64_t           index;
+        LongPollingBatchRequest batch;
+        TStreamId               streamId;
+    };
     std::map<TStreamId, Request>      _requestsByStreamId;
     std::map<TStreamId, ResponseData> _responsesByStreamId;
     std::vector<PendingRequest>       _pendingRequests;
@@ -545,12 +565,90 @@ struct SessionBase {
         self().sendResponse(streamId, code, std::move(response), std::move(extraHeaders));
     }
 
-    void respondWithLongPollingRedirect(TStreamId streamId, const URI<> &topic, std::size_t longPollIdx) {
-        auto location = URI<>::UriFactory(topic).addQueryParameter("LongPollingIdx", std::to_string(longPollIdx)).build();
-        self().respondWithRedirect(streamId, location.str());
+    void respondWithLongPollingRedirect(TStreamId streamId, const URI<> &topic, std::uint64_t longPollIdx, const std::optional<std::string> &batch) {
+        auto factory = URI<>::UriFactory(topic).addQueryParameter("LongPollingIdx", std::to_string(longPollIdx));
+        if (batch.has_value()) {
+            std::move(factory).addQueryParameter(std::string(long_polling::kBatchParameter), *batch);
+        }
+        const auto                location = factory.build();
+
+        majordomo::rest::Response response;
+        response.code = 302;
+        response.headers.emplace_back("location", location.str());
+        response.headers.emplace_back("content-length", "0");
+        self().sendResponse(streamId, std::move(response));
+    }
+
+    void respondToLongPollBatch(TStreamId streamId, const SubscriptionCacheEntry &entry, std::uint64_t firstIndex, std::size_t count) {
+        std::vector<long_polling::BatchPart> parts;
+        parts.reserve(count);
+        const auto firstOffset = firstIndex - entry.firstIndex;
+        for (std::size_t offset = 0; offset < count; ++offset) {
+            const auto &message = entry.messages[firstOffset + offset];
+            const auto  payload = message.error.empty() ? message.data.asString() : std::string_view(message.error);
+            parts.emplace_back(firstIndex + offset, message.topic.str(), message.serviceName, payload);
+        }
+
+        const auto encoded = long_polling::encodeBatch(parts);
+        if (!encoded.has_value()) {
+            respondToLongPollWithError(streamId, encoded.error(), kHttpError, firstIndex);
+            return;
+        }
+
+        majordomo::rest::Response response;
+        response.code = kHttpOk;
+        response.headers.emplace_back("content-type", std::format("multipart/mixed; boundary={}", long_polling::kBoundary));
+        response.headers.emplace_back("content-length", std::to_string(encoded->size()));
+        response.headers.emplace_back("x-opencmw-long-polling-idx", std::to_string(firstIndex));
+        response.body = IoBuffer(encoded->data(), encoded->size());
+        self().sendResponse(streamId, std::move(response));
+    }
+
+    std::expected<LongPollingBatchRequest, std::string> parseLongPollingBatch(const Request &request) const {
+        if (!request.longPollBatch.has_value()) {
+            return LongPollingBatchRequest{};
+        }
+        if (*request.longPollBatch == long_polling::kAllAvailable) {
+            return LongPollingBatchRequest{ .mode = LongPollingBatchRequest::Mode::AllAvailable };
+        }
+
+        const auto parsed = long_polling::parseUnsigned(*request.longPollBatch, long_polling::kBatchParameter);
+        if (!parsed.has_value()) {
+            return std::unexpected(parsed.error());
+        }
+        if (*parsed == 0 || *parsed > SubscriptionCacheEntry::kCapacity) {
+            return std::unexpected(std::format("LongPollingBatch must be between 1 and {}", SubscriptionCacheEntry::kCapacity));
+        }
+        return LongPollingBatchRequest{ .mode = LongPollingBatchRequest::Mode::FixedCount, .count = static_cast<std::size_t>(*parsed) };
     }
 
     std::optional<Message> processLongPollRequest(TStreamId streamId, const Request &request) {
+        std::optional<std::uint64_t> parsedIndex;
+        if (request.longPollIndex != "Next" && request.longPollIndex != "Last") {
+            const auto result = long_polling::parseUnsigned(request.longPollIndex.value_or(""), long_polling::kIndexParameter);
+            if (!result.has_value()) {
+                respondWithError(streamId, result.error(), kHttpBadRequest);
+                return {};
+            }
+            parsedIndex = *result;
+        }
+
+        const auto parsedBatch = parseLongPollingBatch(request);
+        if (!parsedBatch.has_value()) {
+            if (parsedIndex.has_value()) {
+                respondToLongPollWithError(streamId, parsedBatch.error(), kHttpBadRequest, *parsedIndex);
+            } else {
+                respondWithError(streamId, parsedBatch.error(), kHttpBadRequest);
+            }
+            return {};
+        }
+        const auto batch = *parsedBatch;
+        if (parsedIndex.has_value() && batch.mode == LongPollingBatchRequest::Mode::FixedCount
+                && *parsedIndex > std::numeric_limits<std::uint64_t>::max() - (batch.count - 1)) {
+            respondToLongPollWithError(streamId, "LongPollingIdx + LongPollingBatch overflows", kHttpBadRequest, *parsedIndex);
+            return {};
+        }
+
         std::optional<Message> result;
         const auto             zmqTopic = request.topic.toZmqTopic();
         auto                   entryIt  = _sharedData->_subscriptionCache.find(zmqTopic);
@@ -562,19 +660,16 @@ struct SessionBase {
         }
         auto &entry = entryIt->second;
         if (request.longPollIndex == "Next") {
-            respondWithLongPollingRedirect(streamId, request.topic.toMdpTopic(), entry.nextIndex());
+            respondWithLongPollingRedirect(streamId, request.topic.toMdpTopic(), entry.nextIndex(), request.longPollBatch);
             return result;
         } else if (request.longPollIndex == "Last") {
             const std::size_t last = entry.messages.empty() ? entry.nextIndex() : entry.lastIndex();
-            respondWithLongPollingRedirect(streamId, request.topic.toMdpTopic(), last);
+            respondWithLongPollingRedirect(streamId, request.topic.toMdpTopic(), last, request.longPollBatch);
             return result;
         }
 
-        std::uint64_t index = 0;
-        if (auto [ptr, ec] = std::from_chars(request.longPollIndex.data(), request.longPollIndex.data() + request.longPollIndex.size(), index); ec != std::errc()) {
-            respondWithError(streamId, std::format("Malformed LongPollingIdx '{}'", request.longPollIndex));
-            return {};
-        }
+        assert(parsedIndex.has_value());
+        const auto index = *parsedIndex;
 
 #ifdef OPENCMW_PROFILE_HTTP
         if (index % 100 == 0) {
@@ -583,13 +678,32 @@ struct SessionBase {
         }
 #endif
 
-        if (index < entry.firstIndex) {
+        if (batch.mode == LongPollingBatchRequest::Mode::AllAvailable) {
+            if (!entry.messages.empty() && index <= entry.lastIndex()) {
+                const auto first = std::max(index, entry.firstIndex);
+                respondToLongPollBatch(streamId, entry, first, entry.nextIndex() - first);
+            } else {
+                _pendingPolls.push_back(PendingPoll{ zmqTopic, index, batch, streamId });
+            }
+        } else if (batch.mode == LongPollingBatchRequest::Mode::FixedCount) {
+            const auto first = entry.messages.empty() ? index : std::max(index, entry.firstIndex);
+            if (first > std::numeric_limits<std::uint64_t>::max() - (batch.count - 1)) {
+                respondToLongPollWithError(streamId, "LongPollingIdx + LongPollingBatch overflows", kHttpBadRequest, first);
+            } else {
+                const auto lastRequested = first + batch.count - 1;
+                if (entry.messages.empty() || lastRequested > entry.lastIndex()) {
+                    _pendingPolls.push_back(PendingPoll{ zmqTopic, first, batch, streamId });
+                } else {
+                    respondToLongPollBatch(streamId, entry, first, batch.count);
+                }
+            }
+        } else if (index < entry.firstIndex) {
             // index is too old, redirect to the next index
             HTTP_DBG("Server::LongPoll: index {} < firstIndex {}", index, entry.firstIndex);
-            respondWithLongPollingRedirect(streamId, request.topic.toMdpTopic(), entry.nextIndex());
+            respondWithLongPollingRedirect(streamId, request.topic.toMdpTopic(), entry.nextIndex(), {});
         } else if (entry.messages.empty() || index > entry.lastIndex()) {
             // future index, wait for new messages
-            _pendingPolls.emplace_back(zmqTopic, index, streamId);
+            _pendingPolls.push_back(PendingPoll{ zmqTopic, index, batch, streamId });
         } else {
             // we have a message for this index, send it
             respondToLongPoll(streamId, index, Message(entry.messages[index - entry.firstIndex]));
@@ -642,6 +756,8 @@ struct SessionBase {
             for (const auto &[qkey, qvalue] : pathUri.queryParamMap()) {
                 if (qkey == "LongPollingIdx") {
                     request.longPollIndex = qvalue.value_or("");
+                } else if (qkey == long_polling::kBatchParameter) {
+                    request.longPollBatch = qvalue.value_or("");
                 } else if (qkey == "SubscriptionContext") {
                     request.topic           = mdp::Topic::fromMdpTopic(URI<>(qvalue.value_or("")));
                     haveSubscriptionContext = true;
@@ -666,8 +782,8 @@ struct SessionBase {
         }
 
         request.method = parseMethod(method);
-        // Only GET + longPollIndex => LongPoll
-        if (request.method == RestMethod::Get && !request.longPollIndex.empty()) {
+        // LongPollingBatch is only meaningful together with LongPollingIdx.
+        if (request.method == RestMethod::Get && (request.longPollIndex.has_value() || request.longPollBatch.has_value())) {
             request.method = RestMethod::LongPoll;
         }
 
@@ -686,8 +802,8 @@ struct SessionBase {
 
             switch (request.method) {
             case RestMethod::Options:
-               respondToCorsPreflight(streamId);
-               break;
+                respondToCorsPreflight(streamId);
+                break;
             case RestMethod::Get:
             case RestMethod::Post:
                 if (auto m = processGetSetRequest(streamId, request, idGenerator); m.has_value()) {
@@ -712,9 +828,20 @@ struct SessionBase {
     void handleNotification(std::string_view zmqTopic, std::uint64_t index, const Message &msg) {
         auto pollIt = _pendingPolls.begin();
         while (pollIt != _pendingPolls.end()) {
-            const auto &[pendingZmqTopic, pollIndex, streamId] = *pollIt;
-            if (pendingZmqTopic == zmqTopic && index == pollIndex) {
-                respondToLongPoll(streamId, pollIndex, Message(msg));
+            const auto &[pendingZmqTopic, pollIndex, batch, streamId] = *pollIt;
+            const auto lastRequested                                  = batch.mode == LongPollingBatchRequest::Mode::FixedCount ? pollIndex + batch.count - 1 : pollIndex;
+            const bool ready                                          = batch.mode == LongPollingBatchRequest::Mode::None ? index == pollIndex : index >= lastRequested;
+            if (pendingZmqTopic == zmqTopic && ready) {
+                if (batch.mode == LongPollingBatchRequest::Mode::None) {
+                    respondToLongPoll(streamId, pollIndex, Message(msg));
+                } else if (const auto entry = _sharedData->_subscriptionCache.find(pendingZmqTopic); entry != _sharedData->_subscriptionCache.end()) {
+                    if (batch.mode == LongPollingBatchRequest::Mode::AllAvailable) {
+                        const auto first = std::max(pollIndex, entry->second.firstIndex);
+                        respondToLongPollBatch(streamId, entry->second, first, static_cast<std::size_t>(entry->second.nextIndex() - first));
+                    } else {
+                        respondToLongPollBatch(streamId, entry->second, pollIndex, batch.count);
+                    }
+                }
                 pollIt = _pendingPolls.erase(pollIt);
             } else {
                 ++pollIt;
@@ -905,26 +1032,6 @@ struct Http2Session : public SessionBase<Http2Session, std::int32_t> {
         }
     }
 
-    void respondWithRedirect(std::int32_t streamId, std::string_view location) {
-        HTTP_DBG("Server::respondWithRedirect: streamId={} location={}", streamId, location);
-        // :status must go first
-        constexpr auto noCopy  = NGHTTP2_NV_FLAG_NO_COPY_NAME | NGHTTP2_NV_FLAG_NO_COPY_VALUE;
-         auto           headers = std::vector{
-             nv(u8span(":status"), u8span("302"), noCopy),
-             nv(u8span("location"), u8span(location)),
-             nv(u8span("access-control-allow-origin"), u8span("*"), noCopy)};
-
-        if (!_sharedData->_altSvcHeaderValue.empty()) {
-            headers.push_back(_sharedData->_altSvcHeader);
-        }
-        nghttp2_submit_response2(_session, streamId, headers.data(), headers.size(), nullptr);
-    }
-
-    void respondWithLongPollingRedirect(std::int32_t streamId, const URI<> &topic, std::size_t longPollIdx) {
-        auto location = URI<>::UriFactory(topic).addQueryParameter("LongPollingIdx", std::to_string(longPollIdx)).build();
-        respondWithRedirect(streamId, location.str());
-    }
-
     int frame_recv_callback(const nghttp2_frame *frame) {
         HTTP_DBG("Server::Frame: id={} {} {} {}", frame->hd.stream_id, frame->hd.type, frame->hd.flags, (frame->hd.flags & NGHTTP2_FLAG_END_STREAM) ? "END_STREAM" : "");
         switch (frame->hd.type) {
@@ -968,7 +1075,7 @@ struct Http2Session : public SessionBase<Http2Session, std::int32_t> {
         // if this was canceled by the client, remove any pending requests/polls
         if (erased > 0) {
             std::erase_if(_pendingRequests, [stream_id](const auto &request) { return std::get<1>(request) == stream_id; });
-            std::erase_if(_pendingPolls, [stream_id](const auto &poll) { return std::get<2>(poll) == stream_id; });
+            std::erase_if(_pendingPolls, [stream_id](const auto &poll) { return poll.streamId == stream_id; });
         }
         return 0;
     }
@@ -1196,18 +1303,6 @@ struct Http3Session : public SessionBase<Http3Session<TServer>, std::int64_t>, p
         }
     }
 
-    void respondWithRedirect(std::int64_t streamId, std::string_view location) {
-        HTTP_DBG("Server::H3::respondWithRedirect: streamId={} location={}", streamId, location);
-        // :status must go first
-        constexpr auto noCopy  = NGHTTP3_NV_FLAG_NO_COPY_NAME | NGHTTP3_NV_FLAG_NO_COPY_VALUE;
-        const auto     headers = std::array{
-            nv3(u8span(":status"), u8span("302"), noCopy),
-            nv3(u8span("location"), u8span(location)),
-            nv3(u8span("access-control-allow-origin"), u8span("*"), noCopy) };
-
-        nghttp3_conn_submit_response(_httpconn, streamId, headers.data(), headers.size(), nullptr);
-    }
-
     int init(const Endpoint &ep, const Address &local_addr, const sockaddr *sa, socklen_t salen, const ngtcp2_cid *dcid, const ngtcp2_cid *scid, const ngtcp2_cid *ocid, std::span<const std::uint8_t> token, ngtcp2_token_type token_type, std::uint32_t version, TLSServerContext &tls_ctx) {
         auto handshakeCompleted = [](ngtcp2_conn *, void *user_data) {
             auto session = static_cast<Http3Session *>(user_data);
@@ -1381,7 +1476,7 @@ struct Http3Session : public SessionBase<Http3Session<TServer>, std::int64_t>, p
         callbacks.version_negotiation            = ngtcp2_crypto_version_negotiation_cb;
         callbacks.recv_tx_key                    = recvTxKey;
 
-        _scid.datalen = NGTCP2_SV_SCIDLEN;
+        _scid.datalen                            = NGTCP2_SV_SCIDLEN;
         if (generate_secure_random({ _scid.data, _scid.datalen }) != 0) {
             HTTP_DBG("Could not generate connection ID");
             return -1;
@@ -2216,8 +2311,8 @@ inline std::expected<TcpSocket, std::string> createTcpServerSocket(SSL_CTX *ssl_
 struct RestServer {
     TcpSocket                                                                 _tcpServerSocket;
     Http3ServerSocket                                                         _quicServerSocket;
-    SSL_CTX_Ptr                                                               _sslCtxTcp  = SSL_CTX_Ptr(nullptr, SSL_CTX_free);
-    EVP_PKEY_Ptr                                                              _key        = EVP_PKEY_Ptr(nullptr, EVP_PKEY_free);
+    SSL_CTX_Ptr                                                               _sslCtxTcp = SSL_CTX_Ptr(nullptr, SSL_CTX_free);
+    EVP_PKEY_Ptr                                                              _key       = EVP_PKEY_Ptr(nullptr, EVP_PKEY_free);
     std::vector<X509_Ptr>                                                     _cert;
     std::shared_ptr<SharedData>                                               _sharedData = std::make_shared<SharedData>();
     std::map<int, std::unique_ptr<Http2Session>>                              _h2Sessions;
@@ -2495,8 +2590,8 @@ struct RestServer {
             _sharedData->_altSvcHeaderValue = std::format("h3=\":{}\"; ma=86400", port);
             _sharedData->_altSvcHeader      = nv(u8span("alt-svc"), u8span(_sharedData->_altSvcHeaderValue), NGHTTP2_NV_FLAG_NO_COPY_NAME | NGHTTP2_NV_FLAG_NO_COPY_VALUE);
         }
-        _quicServerSocket               = std::move(quicSocket.value());
-        _endpoint.fd                    = _quicServerSocket.fd;
+        _quicServerSocket = std::move(quicSocket.value());
+        _endpoint.fd      = _quicServerSocket.fd;
         return {};
     }
 
@@ -2508,8 +2603,8 @@ struct RestServer {
 
         *p++                                                  = generate_reserved_version(sa, salen, version);
 
-        *p++        = NGTCP2_PROTO_VER_V1;
-        *p++        = NGTCP2_PROTO_VER_V2;
+        *p++                                                  = NGTCP2_PROTO_VER_V1;
+        *p++                                                  = NGTCP2_PROTO_VER_V2;
 
         auto nwrite                                           = ngtcp2_pkt_write_version_negotiation(buf.wpos(), buf.left(), std::uniform_int_distribution<uint8_t>()(_randgen), dcid.data(), dcid.size(), scid.data(), scid.size(), sv.data(), static_cast<std::size_t>(p - std::begin(sv)));
         if (nwrite < 0) {
diff --git a/src/majordomo/test/majordomoworker_rest_tests.cpp b/src/majordomo/test/majordomoworker_rest_tests.cpp
index 7cddfa21..59e74a6e 100644
--- a/src/majordomo/test/majordomoworker_rest_tests.cpp
+++ b/src/majordomo/test/majordomoworker_rest_tests.cpp
@@ -1,5 +1,6 @@
 #include "majordomo/Rest.hpp"
 #include <majordomo/Broker.hpp>
+#include <majordomo/RestServer.hpp>
 #include <majordomo/Settings.hpp>
 #include <majordomo/Worker.hpp>
 
@@ -477,7 +478,7 @@ TEST_CASE("Subscriptions", "[majordomo][majordomoworker][subscription]") {
 
     opencmw::client::Command                allSub;
     allSub.command  = mdp::Command::Subscribe;
-    allSub.topic    = opencmw::URI<>(std::format("http://localhost:{}/colors", kServerPort));
+    allSub.topic    = opencmw::URI<>(std::format("http://localhost:{}/colors?LongPollingBatch=AllAvailable", kServerPort));
     allSub.callback = [&allReceived, &allExpected](const auto &msg) {
         REQUIRE(msg.command == mdp::Command::Notify);
         REQUIRE(msg.error == "");
@@ -776,3 +777,262 @@ TEST_CASE("Subscription latencies", "[majordomo][majordomoworker][rest]") {
     REQUIRE(nReceived > 10);
     REQUIRE(static_cast<double>(msLatency) / nReceived < 20000); // unit is µs
 }
+
+namespace batch_tests {
+
+using namespace opencmw;
+using namespace opencmw::majordomo::detail::rest;
+
+namespace {
+
+struct CapturedMessageResponse {
+    int                                              code;
+    Message                                          message;
+    std::vector<std::pair<std::string, std::string>> headers;
+};
+
+struct TestSession : SessionBase<TestSession, int> {
+    using SessionBase::SessionBase;
+
+    std::vector<majordomo::rest::Response> responses;
+    std::vector<CapturedMessageResponse>   messageResponses;
+
+    void sendResponse(int, majordomo::rest::Response response) {
+        responses.push_back(std::move(response));
+    }
+
+    void sendResponse(int, int code, Message &&message, const std::vector<nghttp2_nv> &headers = {}) {
+        std::vector<std::pair<std::string, std::string>> copiedHeaders;
+        copiedHeaders.reserve(headers.size());
+        for (const auto &header : headers) {
+            copiedHeaders.emplace_back(
+                    std::string(reinterpret_cast<const char *>(header.name), header.namelen),
+                    std::string(reinterpret_cast<const char *>(header.value), header.valuelen));
+        }
+        messageResponses.push_back({ code, std::move(message), std::move(copiedHeaders) });
+    }
+};
+
+std::string_view header(const std::vector<std::pair<std::string, std::string>> &headers, std::string_view name) {
+    const auto entry = std::ranges::find(headers, name, &std::pair<std::string, std::string>::first);
+    return entry == headers.end() ? std::string_view{} : entry->second;
+}
+
+Message notification(std::uint64_t index) {
+    Message message{};
+    message.command     = mdp::Command::Notify;
+    message.serviceName = std::format("/batch-service-{}", index);
+    message.topic       = URI<>(std::format("/batch?sample={}", index));
+    const auto payload  = std::format("message-{}", index);
+    message.data        = IoBuffer(payload.data(), payload.size());
+    return message;
+}
+
+Request request(std::string index, std::optional<std::string> batch = {}) {
+    Request result;
+    result.topic         = mdp::Topic::fromMdpTopic(URI<>("/batch"));
+    result.method        = RestMethod::LongPoll;
+    result.longPollIndex = std::move(index);
+    result.longPollBatch = std::move(batch);
+    return result;
+}
+
+} // namespace
+
+TEST_CASE("Fixed-count batch: wait for all messages", "[majordomo][rest][long-polling][batch]") {
+    const std::string binaryPayload = std::string(1, '\0') + "\r\n--" + std::string(long_polling::kBoundary);
+
+    auto        shared = std::make_shared<SharedData>();
+    TestSession session(shared);
+
+    REQUIRE(session.processLongPollRequest(1, request("0", "3")).has_value());
+    REQUIRE(session._pendingPolls.size() == 1);
+
+    auto &entry = shared->_subscriptionCache.at("/batch#");
+    for (std::uint64_t index = 0; index < 2; ++index) {
+        auto message = notification(index);
+        if (index == 1) {
+            message.data = IoBuffer(binaryPayload.data(), binaryPayload.size());
+        }
+        entry.add(std::move(message));
+        session.handleNotification("/batch#", index, entry.messages.back());
+        REQUIRE(session.responses.empty());
+    }
+
+    auto errorNotification  = notification(2);
+    errorNotification.error = "error-2";
+    entry.add(std::move(errorNotification));
+    session.handleNotification("/batch#", 2, entry.messages.back());
+
+    REQUIRE(session._pendingPolls.empty());
+    REQUIRE(session.responses.size() == 1);
+    CHECK(header(session.responses.front().headers, "x-opencmw-topic").empty());
+    CHECK(header(session.responses.front().headers, "x-opencmw-service-name").empty());
+    CHECK(header(session.responses.front().headers, "content-type") == std::format("multipart/mixed; boundary={}", long_polling::kBoundary));
+    CHECK(session.responses.front().body.asString().starts_with(std::format("--{}\r\n", long_polling::kBoundary)));
+    CHECK(session.responses.front().body.asString().ends_with(std::format("--{}--\r\n", long_polling::kBoundary)));
+    const auto parts = long_polling::decodeBatch(session.responses.front().body.asString());
+    REQUIRE(parts.has_value());
+    REQUIRE(parts->size() == 3);
+    for (std::uint64_t index = 0; index < parts->size(); ++index) {
+        CHECK((*parts)[index].index == index);
+        CHECK((*parts)[index].topic == std::format("/batch?sample={}", index));
+        CHECK((*parts)[index].serviceName == std::format("/batch-service-{}", index));
+    }
+    CHECK((*parts)[0].payload == "message-0");
+    CHECK((*parts)[1].payload == binaryPayload);
+    CHECK((*parts)[2].payload == "error-2");
+}
+
+TEST_CASE("Batch start: requested or oldest buffered index", "[majordomo][rest][long-polling][batch]") {
+    // requestedIndex 0, AllAvailable -> return 5–104
+    // requestedIndex 7, AllAvailable -> return 7–104
+    // requestedIndex 0, fixed-count 5 -> return 5–9
+    // requestedIndex 7, fixed count 5 -> return 7–11
+    const auto requestedIndex = GENERATE(0, 7);
+    const auto batch          = GENERATE(as<std::string>{}, "AllAvailable", "5");
+    auto       shared         = std::make_shared<SharedData>();
+    auto      &entry          = shared->_subscriptionCache["/batch#"];
+    for (std::uint64_t index = 0; index < SubscriptionCacheEntry::kCapacity + 5; ++index) {
+        entry.add(notification(index));
+    }
+    REQUIRE(entry.firstIndex == 5);
+
+    TestSession session(shared);
+    REQUIRE_FALSE(session.processLongPollRequest(1, request(std::to_string(requestedIndex), batch)).has_value());
+    REQUIRE(session.responses.size() == 1);
+
+    const auto parts = long_polling::decodeBatch(session.responses.front().body.asString());
+    REQUIRE(parts.has_value());
+    const auto first = requestedIndex == 0 ? 5 : 7;
+    const auto count = batch == long_polling::kAllAvailable ? SubscriptionCacheEntry::kCapacity + 5 - first : 5;
+    REQUIRE(parts->size() == count);
+    CHECK(parts->front().index == first);
+    CHECK(parts->back().index == first + count - 1);
+    CHECK(header(session.responses.front().headers, "x-opencmw-long-polling-idx") == std::to_string(first));
+    CHECK(session.messageResponses.empty());
+}
+
+TEST_CASE("Fixed-count batch while old messages are removed", "[majordomo][rest][long-polling][batch]") {
+    // Count 3, request 99-101:
+    //   Initially:           buffer 0-99  -> wait
+    //   Message 100 arrives: buffer 1-100 -> still wait
+    //   Message 101 arrives: buffer 2-101 -> return only 99-101
+    //
+    // Count 100, request 99-198:
+    //   Wait for message 198: buffer 99-198 -> return 99-198 (the entire buffer).
+    const auto count = GENERATE(std::size_t{ 3 }, SubscriptionCacheEntry::kCapacity);
+    CAPTURE(count);
+    auto  shared = std::make_shared<SharedData>();
+    auto &entry  = shared->_subscriptionCache["/batch#"];
+    for (std::uint64_t index = 0; index < SubscriptionCacheEntry::kCapacity; ++index) {
+        entry.add(notification(index));
+    }
+
+    const auto  first = entry.lastIndex();
+    const auto  last  = first + count - 1;
+    TestSession session(shared);
+    REQUIRE_FALSE(session.processLongPollRequest(1, request(std::to_string(first), std::to_string(count))).has_value());
+    REQUIRE(session._pendingPolls.size() == 1);
+    REQUIRE(session.responses.empty());
+
+    for (auto index = first + 1; index <= last; ++index) {
+        entry.add(notification(index));
+        session.handleNotification("/batch#", index, entry.messages.back());
+        if (index < last) {
+            REQUIRE(session.responses.empty());
+        }
+    }
+
+    CHECK(entry.firstIndex == count - 1);
+    CHECK(session._pendingPolls.empty());
+    CHECK(session.messageResponses.empty());
+    REQUIRE(session.responses.size() == 1);
+    const auto parts = long_polling::decodeBatch(session.responses.front().body.asString());
+    REQUIRE(parts.has_value());
+    REQUIRE(parts->size() == count);
+    CHECK(parts->front().index == first);
+    CHECK(parts->back().index == last);
+    for (const auto &part : *parts) {
+        CHECK(part.payload == std::format("message-{}", part.index));
+    }
+}
+
+TEST_CASE("AllAvailable: wait for a future index", "[majordomo][rest][long-polling][batch]") {
+    auto        shared = std::make_shared<SharedData>();
+    TestSession session(shared);
+    REQUIRE(session.processLongPollRequest(1, request("1", "AllAvailable")).has_value());
+    REQUIRE(session._pendingPolls.size() == 1);
+    CHECK(session.responses.empty());
+
+    auto &entry = shared->_subscriptionCache.at("/batch#");
+    entry.add(notification(0));
+    session.handleNotification("/batch#", 0, entry.messages.back());
+    CHECK(session.responses.empty());
+
+    entry.add(notification(1));
+    session.handleNotification("/batch#", 1, entry.messages.back());
+    CHECK(session._pendingPolls.empty());
+    REQUIRE(session.responses.size() == 1);
+    const auto parts = long_polling::decodeBatch(session.responses.front().body.asString());
+    REQUIRE(parts.has_value());
+    REQUIRE(parts->size() == 1);
+    CHECK(parts->front().index == 1);
+}
+
+TEST_CASE("Next redirect keeps LongPollingBatch", "[majordomo][rest][long-polling][batch]") {
+    auto        shared = std::make_shared<SharedData>();
+    TestSession session(shared);
+
+    REQUIRE(session.processLongPollRequest(1, request("Next", "AllAvailable")).has_value());
+    REQUIRE(session.responses.size() == 1);
+    CHECK(session.responses.front().code == 302);
+
+    const URI<> location{ std::string(header(session.responses.front().headers, "location")) };
+    const auto &parameters = location.queryParamMap();
+    CHECK(parameters.at("LongPollingIdx") == "0");
+    CHECK(parameters.at("LongPollingBatch") == "AllAvailable");
+}
+
+TEST_CASE("Invalid long-poll requests: HTTP 400", "[majordomo][rest][long-polling][batch]") {
+    const auto query = GENERATE(as<std::string>{},
+            "LongPollingIdx=invalid",
+            "LongPollingIdx=1x",
+            "LongPollingIdx",
+            "LongPollingIdx=",
+            "LongPollingBatch=5",
+            "LongPollingIdx=Next&LongPollingBatch=0",
+            "LongPollingIdx=0&LongPollingBatch=0",
+            "LongPollingIdx=0&LongPollingBatch",
+            "LongPollingIdx=0&LongPollingBatch=",
+            "LongPollingIdx=0&LongPollingBatch=2x",
+            "LongPollingIdx=0&LongPollingBatch=101",
+            "LongPollingIdx=18446744073709551615&LongPollingBatch=2");
+    CAPTURE(query);
+    auto        shared = std::make_shared<SharedData>();
+    TestSession session(shared);
+
+    const auto  process = [&](int streamId, std::string path) {
+        session.addHeader(streamId, ":method", "GET");
+        session.addHeader(streamId, ":path", path);
+        session.processCompletedRequest(streamId);
+
+        IdGenerator idGenerator;
+        return session.getMessages(idGenerator);
+    };
+
+    CHECK(process(1, std::format("/batch?{}", query)).empty());
+    REQUIRE(shared->_subscriptionCache.empty());
+    CHECK(session._pendingPolls.empty());
+    CHECK(session.responses.empty());
+    REQUIRE(session.messageResponses.size() == 1);
+    CHECK(session.messageResponses.front().code == kHttpBadRequest);
+
+    const auto messages = process(2, "/batch?LongPollingIdx=0&LongPollingBatch=1");
+    REQUIRE(messages.size() == 1);
+    CHECK(messages.front().command == mdp::Command::Subscribe);
+    CHECK(shared->_subscriptionCache.contains("/batch#"));
+    CHECK(session._pendingPolls.size() == 1);
+}
+
+} // namespace batch_tests