Repository: ndsev/mapget (language: python)

Issue:
Requests to unknown map+layer combinations receive a 400 response
When a fetch request is received for a map+layer combination for which no connected datasource can provide the response, a 400 response should be given with a descriptive error message. 


Discussion / hints:

Fix patch:
diff --git a/README.md b/README.md
index 348c4766..61cf0daf 100644
--- a/README.md
+++ b/README.md
@@ -251,7 +251,7 @@ void main(int argc, char const *argv[])
      HttpClient client("localhost", service.port());
 
      auto receivedTileCount = 0;
-     client.request(std::make_shared<Request>(
+     client.request(std::make_shared<LayerTilesRequest>(
          "Tropico",
          "WayLayer",
          std::vector<TileId>{{1234, 5678, 9112, 1234}},
@@ -264,3 +264,16 @@ void main(int argc, char const *argv[])
 ```
 
 Keep in mind, that you can also run a `mapget` service without any RPCs in your application. Check out [`examples/cpp/local-datasource`](examples/cpp/local-datasource/main.cpp) on how to do that.
+
+### erdblick-mapget-datasource communication pattern
+
+TODO: expand and polish this section stub.
+
+1. Client (`erdblick` etc.) sends a composite list of requests to `mapget`. Requests are batched because browsers limit the number of concurrent requests to one domain, but we want to stream potentially hundreds of tiles.
+
+2. `mapget` checks if all requested map+layer combinations can be fulfilled with data sources
+   - yes: create tile requests, stream responses back to client,
+   - no: return 400 Bad Request (client needs to refresh its info on map availability).
+
+3. A data source drops offline / `mapget` request fails during processing?
+   - `cpp-httplib` cleanup callback returns timeout response (probably status code 408).
diff --git a/examples/cpp/local-datasource/main.cpp b/examples/cpp/local-datasource/main.cpp
index 83f79a04..7070d7f2 100644
--- a/examples/cpp/local-datasource/main.cpp
+++ b/examples/cpp/local-datasource/main.cpp
@@ -76,13 +76,13 @@ int main(int argc, char** argv)
     service.add(std::make_shared<MyLocalDataSource>());
 
     // Create a request. The request will immediately start to be worked on.
-    service
-        .request(std::make_shared<Request>(
+    auto r = std::make_shared<LayerTilesRequest>(
             "Tropico",
             "WayLayer",
             std::vector<TileId>{TileId(12345), TileId(67689)},
             [](auto&& result)
-            { log().info("Got {}", MapTileKey(*result).toString()); }))
-        ->wait();
+            { log().info("Got {}", MapTileKey(*result).toString()); });
+    service.request({r});
+    r->wait();
     return 0;
 }
diff --git a/libs/http-datasource/include/mapget/http-datasource/datasource-server.h b/libs/http-datasource/include/mapget/http-datasource/datasource-server.h
index 423bff6c..afbe3c09 100644
--- a/libs/http-datasource/include/mapget/http-datasource/datasource-server.h
+++ b/libs/http-datasource/include/mapget/http-datasource/datasource-server.h
@@ -24,11 +24,11 @@ class DataSourceServer : public HttpServer
     ~DataSourceServer() override;
 
     /**
-     * Set the Callback which will be invoked when a `/tile`-request is received.
+     * Set the callback which will be invoked when a `/tile`-request is received.
      * The callback argument is a fresh TileFeatureLayer, which the callback
      * must fill according to the set TileFeatureLayer's layer info and tile id.
      * If an error occurs while filling the tile, the callback can use
-     * TileFeatureLayer::setError(...) to singal the error downstream.
+     * TileFeatureLayer::setError(...) to signal the error downstream.
      */
     DataSourceServer& onTileRequest(std::function<void(TileFeatureLayer::Ptr)> const&);
 
diff --git a/libs/http-datasource/src/datasource-server.cpp b/libs/http-datasource/src/datasource-server.cpp
index 8f3b5751..5567b0b1 100644
--- a/libs/http-datasource/src/datasource-server.cpp
+++ b/libs/http-datasource/src/datasource-server.cpp
@@ -44,6 +44,8 @@ void DataSourceServer::setup(httplib::Server& server)
     server.Get(
         "/tile",
         [this](const httplib::Request& req, httplib::Response& res) {
+
+            // Extract parameters from request.
             auto layerIdParam = req.get_param_value("layer");
             auto layer = impl_->info_.getLayer(layerIdParam);
 
@@ -57,6 +59,7 @@ void DataSourceServer::setup(httplib::Server& server)
             if (req.has_param("responseType"))
                 responseType = req.get_param_value("responseType");
 
+            // Create response TileFeatureLayer.
             auto tileFeatureLayer = std::make_shared<TileFeatureLayer>(
                 tileIdParam,
                 impl_->info_.nodeId_,
@@ -66,7 +69,7 @@ void DataSourceServer::setup(httplib::Server& server)
             if (impl_->tileCallback_)
                 impl_->tileCallback_(tileFeatureLayer);
 
-            // Serialize TileFeatureLayer using TileLayerStream
+            // Serialize TileFeatureLayer using TileLayerStream.
             if (responseType == "binary") {
                 std::stringstream content;
                 TileLayerStream::FieldOffsetMap fieldOffsets{
diff --git a/libs/http-service/include/mapget/http-service/http-client.h b/libs/http-service/include/mapget/http-service/http-client.h
index 5936bf5c..e0f8c450 100644
--- a/libs/http-service/include/mapget/http-service/http-client.h
+++ b/libs/http-service/include/mapget/http-service/http-client.h
@@ -7,7 +7,9 @@
 
 namespace mapget {
 
-/** Client class, which implements asynchronous fetching from a mapget HTTP service.*/
+/**
+ * Client class, which implements asynchronous fetching from a mapget HTTP service.
+ */
 class HttpClient
 {
 public:
@@ -26,7 +28,7 @@ class HttpClient
     /**
      * Post a Request for a number of tiles from a particular map layer.
      */
-    Request::Ptr request(Request::Ptr const& request);
+    LayerTilesRequest::Ptr request(LayerTilesRequest::Ptr const& request);
 
 private:
     struct Impl;
diff --git a/libs/http-service/src/cli.cpp b/libs/http-service/src/cli.cpp
index 90ef4e07..f554ba6c 100644
--- a/libs/http-service/src/cli.cpp
+++ b/libs/http-service/src/cli.cpp
@@ -107,7 +107,7 @@ struct FetchCommand
         int port = std::stoi(server_.substr(delimiterPos+1, server_.size()));
 
         mapget::HttpClient cli(host, port);
-        auto request = std::make_shared<Request>(
+        auto request = std::make_shared<LayerTilesRequest>(
             map_, layer_, std::vector<TileId>{tiles_.begin(), tiles_.end()},
             [](auto&& tile){
                 if (log().level() == spdlog::level::trace) {
diff --git a/libs/http-service/src/http-client.cpp b/libs/http-service/src/http-client.cpp
index ee9144d9..5d2c354d 100644
--- a/libs/http-service/src/http-client.cpp
+++ b/libs/http-service/src/http-client.cpp
@@ -28,7 +28,7 @@ struct HttpClient::Impl {
     {
         auto mapIt = sources_.find(std::string(map));
         if (mapIt == sources_.end())
-            throw logRuntimeError("Could nto find map data source info");
+            throw logRuntimeError("Could not find map data source info");
         return mapIt->second.getLayer(std::string(layer));
     }
 };
@@ -45,10 +45,13 @@ std::vector<DataSourceInfo> HttpClient::sources() const
     return result;
 }
 
-Request::Ptr HttpClient::request(const Request::Ptr& request)
+LayerTilesRequest::Ptr HttpClient::request(const LayerTilesRequest::Ptr& request)
 {
-    if (request->isDone())
-        request->notifyDone();
+    // Finalize requests that did not contain any tiles.
+    if (request->isDone()) {
+        request->notifyStatus();
+        return request;
+    }
 
     auto reader = std::make_unique<TileLayerStream::Reader>(
         [this](auto&& mapId, auto&& layerId){return impl_->resolve(mapId, layerId);},
@@ -64,8 +67,16 @@ Request::Ptr HttpClient::request(const Request::Ptr& request)
         json::object({{"requests", json::array({request->toJson()})}}).dump(),
         "application/json");
 
-    if (tileResponse && tileResponse->status == 200)
-        reader->read(tileResponse->body);
+    if (tileResponse) {
+        if (tileResponse->status == 200) {
+            reader->read(tileResponse->body);
+        }
+        else if (tileResponse->status == 400) {
+            request->setStatus(mapget::RequestStatus::NoDataSource);
+        }
+        // TODO if multiple LayerTileRequests are ever sent by this client,
+        //  additionally handle RequestStatus::Aborted.
+    }
 
     return request;
 }
diff --git a/libs/http-service/src/http-service.cpp b/libs/http-service/src/http-service.cpp
index 662b4889..9e49de88 100644
--- a/libs/http-service/src/http-service.cpp
+++ b/libs/http-service/src/http-service.cpp
@@ -12,8 +12,8 @@ struct HttpService::Impl
 
     explicit Impl(HttpService& self) : self_(self) {}
 
-    // Use a shared buffer for the responses and a mutex for thread safety
-    struct TileLayerRequestState
+    // Use a shared buffer for the responses and a mutex for thread safety.
+    struct HttpTilesRequestState
     {
         static constexpr auto binaryMimeType = "application/binary";
         static constexpr auto jsonlMimeType = "application/jsonl";
@@ -24,10 +24,10 @@ struct HttpService::Impl
 
         std::stringstream buffer_;
         std::string responseType_;
-        std::vector<Request::Ptr> requests_;
+        std::vector<LayerTilesRequest::Ptr> requests_;
         TileLayerStream::FieldOffsetMap fieldsOffsets_;
 
-        void addRequestFromJson(nlohmann::json const& requestJson)
+        void parseRequestFromJson(nlohmann::json const& requestJson)
         {
             std::string mapId = requestJson["mapId"];
             std::string layerId = requestJson["layerId"];
@@ -36,17 +36,17 @@ struct HttpService::Impl
             for (auto const& tid : requestJson["tileIds"].get<std::vector<uint64_t>>())
                 tileIds.emplace_back(tid);
             requests_.push_back(
-                std::make_shared<Request>(mapId, layerId, std::move(tileIds), nullptr));
+                std::make_shared<LayerTilesRequest>(mapId, layerId, std::move(tileIds), nullptr));
         }
 
         void setResponseType(std::string const& s)
         {
             responseType_ = s;
-            if (responseType_ == TileLayerRequestState::binaryMimeType)
+            if (responseType_ == HttpTilesRequestState::binaryMimeType)
                 return;
-            if (responseType_ == TileLayerRequestState::jsonlMimeType)
+            if (responseType_ == HttpTilesRequestState::jsonlMimeType)
                 return;
-            if (responseType_ == TileLayerRequestState::anyMimeType) {
+            if (responseType_ == HttpTilesRequestState::anyMimeType) {
                 responseType_ = binaryMimeType;
                 return;
             }
@@ -71,51 +71,76 @@ struct HttpService::Impl
         }
     };
 
+    /**
+     * Wraps around the generic mapget service's request() function
+     * to include httplib request decoding and response encoding.
+     */
     void handleTilesRequest(const httplib::Request& req, httplib::Response& res)
     {
-        // Parse the JSON request
+        // Parse the JSON request.
         nlohmann::json j = nlohmann::json::parse(req.body);
         auto requestsJson = j["requests"];
-        // TODO: Sanity-check length of requests-array. The user is expected
-        //  to create one map-layer request object per map-layer combination.
-        //  Pumping up the number of nested per-map-layer requests (extreme: one map-layer
-        //  request object per tile) would be a great way to stall the server for other users.
-        auto state = std::make_shared<TileLayerRequestState>();
+
+        // TODO: Limit number of requests to avoid DoS to other users.
+        // Within one HTTP request, all requested tiles from the same map+layer
+        // combination should be in a single LayerTilesRequest.
+        auto state = std::make_shared<HttpTilesRequestState>();
         for (auto& requestJson : requestsJson) {
-            state->addRequestFromJson(requestJson);
+            state->parseRequestFromJson(requestJson);
         }
 
-        // Parse maxKnownFieldIds
+        // Parse maxKnownFieldIds.
         if (j.contains("maxKnownFieldIds")) {
             for (auto& item : j["maxKnownFieldIds"].items()) {
                 state->fieldsOffsets_[item.key()] = item.value().get<simfil::FieldId>();
             }
         }
 
-        // Determine response type
+        // Determine response type.
         state->setResponseType(req.get_header_value("Accept"));
 
-        // Process requests
+        // Process requests.
         for (auto& request : state->requests_) {
             request->onResult_ = [state](auto&& tileFeatureLayer)
             {
                 state->addResult(tileFeatureLayer);
             };
-            request->onDone_ = [state]
+            request->onDone_ = [state](RequestStatus r)
             {
                 state->resultEvent_.notify_one();
             };
-            self_.request(request);
+        }
+        auto canProcess = self_.request(state->requests_);
+
+        if (!canProcess) {
+            // Send a status report detailing for each request
+            // whether its data source is unavailable or it was aborted.
+            res.status = 400;
+            std::vector<int> requestStatuses{};
+            for (const auto& r : state->requests_) {
+                requestStatuses.push_back(r->getStatus());
+            }
+            res.set_content(
+                nlohmann::json::object({{"requestStatuses", requestStatuses}}).dump(),
+                "application/json");
+            return;
         }
 
-        // Set up the streaming response
+        // For efficiency, set up httplib to stream tile layer responses to client:
+        // (1) Lambda continuously supplies response data to httplib's DataSink,
+        //     picking up data from state->buffer_ until all tile requests are done.
+        //     Then, signal sink->done() to close the stream with a 200 status.
+        //     See httplib::write_content_without_length(...) too.
+        // (2) Lambda acts as a cleanup routine, triggered by httplib upon request wrap-up.
+        //     The success flag indicates if wrap-up was due to sink->done() or external factors
+        //     like network errors or request aborts in lengthy tile requests (e.g., map-viewer).
         res.set_content_provider(
             state->responseType_,
             [state](size_t offset, httplib::DataSink& sink)
             {
                 std::unique_lock lock(state->mutex_);
 
-                // Wait until there is data to be read
+                // Wait until there is data to be read.
                 std::string strBuf;
                 bool allDone = false;
                 state->resultEvent_.wait(
@@ -133,17 +158,18 @@ struct HttpService::Impl
                 if (!strBuf.empty()) {
                     log().debug("Streaming {} bytes...", strBuf.size());
                     sink.write(strBuf.data(), strBuf.size());
-                    state->buffer_.str("");  // Clear buffer after reading
+                    state->buffer_.str("");  // Clear buffer after reading.
                 }
 
-                // Call sink.done() when all requests are done
+                // Call sink.done() when all requests are done.
                 if (allDone) {
                     sink.done();
                 }
 
                 return true;
             },
-            // Cleanup callback to abort the requests
+            // Network error/timeout of request to datasource:
+            // cleanup callback to abort the requests.
             [state, this](bool success)
             {
                 if (!success) {
diff --git a/libs/model/include/mapget/model/layer.h b/libs/model/include/mapget/model/layer.h
index f4a8f3b4..e9673ba0 100644
--- a/libs/model/include/mapget/model/layer.h
+++ b/libs/model/include/mapget/model/layer.h
@@ -153,6 +153,7 @@ class TileLayer
     std::string nodeId_;
     std::string mapId_;
     std::shared_ptr<LayerInfo> layerInfo_;
+    // TODO what is the error string for? Does it follow some standard?
     std::optional<std::string> error_;
     std::chrono::time_point<std::chrono::system_clock> timestamp_;
     std::optional<std::chrono::milliseconds> ttl_;
diff --git a/libs/model/src/info.cpp b/libs/model/src/info.cpp
index bebcb87b..ee2420b8 100644
--- a/libs/model/src/info.cpp
+++ b/libs/model/src/info.cpp
@@ -214,7 +214,7 @@ std::shared_ptr<LayerInfo> DataSourceInfo::getLayer(std::string const& layerId)
     auto it = layers_.find(layerId);
     if (it == layers_.end()) {
         throw logRuntimeError(
-            stx::format("Could not find info for layer '{}'/'{}'", mapId_, layerId)
+            stx::format("Could not find layer '{}' in map '{}'", layerId, mapId_)
         );
     }
     return it->second;
diff --git a/libs/pymapget/binding/py-client.h b/libs/pymapget/binding/py-client.h
index d498e3ad..4f9f77b6 100644
--- a/libs/pymapget/binding/py-client.h
+++ b/libs/pymapget/binding/py-client.h
@@ -60,20 +60,22 @@ namespace mapget
 
 namespace py = pybind11;
 
-class PyRequest : public Request {
+class PyRequest : public LayerTilesRequest
+{
 public:
-    using Request::Request;
+    using LayerTilesRequest::LayerTilesRequest;
 
     void notifyResult(TileFeatureLayer::Ptr result) override {
         std::unique_lock lock(bufferMutex_);
         buffer_.push(result);
         bufferSignal_.notify_one();  // Signal that a new result is available
-        Request::notifyResult(result);
+        LayerTilesRequest::notifyResult(result);
     }
 
     TileFeatureLayer::Ptr next() {
         std::unique_lock lock(bufferMutex_);
-        bufferSignal_.wait(lock, [this](){ return !buffer_.empty() || isDone(); });
+        bufferSignal_.wait(lock, [this](){ return !buffer_.empty() ||
+                                 this->getStatus() != RequestStatus::Open; });
         if (buffer_.empty()) {
             throw py::stop_iteration();
         } else {
diff --git a/libs/service/include/mapget/service/datasource.h b/libs/service/include/mapget/service/datasource.h
index d58e9770..86716d1b 100644
--- a/libs/service/include/mapget/service/datasource.h
+++ b/libs/service/include/mapget/service/datasource.h
@@ -28,7 +28,7 @@ class DataSource
      * times in parallel to satisfy data requests for a mapget Service.
      * @param featureTile A TileFeatureLayer object which this data source
      *  should fill according the available data. If any error occurs
-     *  while dong so, the data source may use TileLayer::setError.
+     *  while doing so, the data source may use TileLayer::setError.
      *  To store any extra information of interest such as timings or sizes,
      *  TileLayer::setInfo() may be used.
      */
diff --git a/libs/service/include/mapget/service/service.h b/libs/service/include/mapget/service/service.h
index abaf679f..1cf6341f 100644
--- a/libs/service/include/mapget/service/service.h
+++ b/libs/service/include/mapget/service/service.h
@@ -10,32 +10,42 @@
 namespace mapget
 {
 
+enum RequestStatus {
+    Open = 0x0,
+    Success = 0x1, /** The request has been fully satisfied. */
+    NoDataSource = 0x2, /** No data source could provide the requested map + layer. */
+    Aborted = 0x3 /** Canceled, e.g. because a bundled request cannot be fulfilled. */
+};
+
 /**
  * Client request for map data, which consists of a map id,
  * a map layer id, an array of tile ids, and a callback function
  * which signals results.
  */
-class Request
+class LayerTilesRequest
 {
     friend class Service;
     friend class HttpClient;
 
 public:
-    using Ptr = std::shared_ptr<Request>;
+    using Ptr = std::shared_ptr<LayerTilesRequest>;
 
     /** Construct a request with the relevant parameters. */
-    Request(
+    LayerTilesRequest(
         std::string mapId,
         std::string layerId,
         std::vector<TileId> tiles,
         std::function<void(TileFeatureLayer::Ptr)> onResult);
 
-    /** Check if the request has been fully satisfied. */
-    [[nodiscard]] bool isDone() const;
+    /** Get the current status of the request. */
+    RequestStatus getStatus();
 
     /** Wait for the request to be done. */
     void wait();
 
+    /** Check whether the request is done or still running. */
+    bool isDone();
+
     /** The map id for which this request is dedicated. */
     std::string mapId_;
 
@@ -44,7 +54,7 @@ class Request
 
     /**
      * The map tile ids for which this request is dedicated.
-     * Must not be empty. Result tiles will be processed in the given order
+     * Must not be empty. Result tiles will be processed in the given order.
      */
     std::vector<TileId> tiles_;
 
@@ -56,13 +66,15 @@ class Request
     /**
      * The callback function which is called when all tiles have been processed.
      */
-    std::function<void()> onDone_;
+    std::function<void(RequestStatus)> onDone_;
 
 protected:
     virtual void notifyResult(TileFeatureLayer::Ptr);
-    void notifyDone();
+    void setStatus(RequestStatus s);
+    void notifyStatus();
     nlohmann::json toJson();
 
+private:
     // So the service can track which tileId index from tiles_
     // is next in line to be processed.
     size_t nextTileIndex_ = 0;
@@ -70,9 +82,10 @@ class Request
     // So the requester can track how many results have been received.
     size_t resultCount_ = 0;
 
-    // Mutex/Condition variable for done-signal
-    std::mutex doneMutex_;
-    std::condition_variable doneConditionVariable_;
+    // Mutex/condition variable for reading/setting request status.
+    std::mutex statusMutex_;
+    std::condition_variable statusConditionVariable_;
+    RequestStatus status_ = Open;
 };
 
 /**
@@ -110,23 +123,27 @@ class Service
     void remove(DataSource::Ptr const& dataSource);
 
     /**
-     * Request some map data tiles. Will throw an exception if
-     * there is no worker which is able to process the request.
-     * Note: The same request object should only ever be passed
-     *  to one service. Otherwise, there is undefined behavior.
-     * @return The `r` parameter value is returned.
+     * Request some map data tiles. If the requested map+layer
+     * combination is available, will schedule a job to retrieve
+     * the tiles. A request object should only ever be passed
+     * to one service. Otherwise, there is undefined behavior.
+     * @return false if the requested map+layer is not available
+     * from any connected DataSource, true otherwise.
      */
-    Request::Ptr request(Request::Ptr r);
+    bool request(std::vector<LayerTilesRequest::Ptr> requests);
 
     /**
      * Abort the given request. The request will be removed from
      * the processing queue, and forcefully marked as done.
      */
-    void abort(Request::Ptr const& r);
+    void abort(LayerTilesRequest::Ptr const& r);
 
     /** DataSourceInfo for all data sources which have been added to this Service. */
     std::vector<DataSourceInfo> info();
 
+    /** Checks if a DataSource can serve the requested map+layer combination. */
+    bool hasLayer(std::string const& mapId, std::string const& layerId);
+
     /** Get the Cache which this service was constructed with. */
     [[nodiscard]] Cache::Ptr cache();
 
diff --git a/libs/service/src/service.cpp b/libs/service/src/service.cpp
index cc789021..a69059ca 100644
--- a/libs/service/src/service.cpp
+++ b/libs/service/src/service.cpp
@@ -11,7 +11,7 @@
 namespace mapget
 {
 
-Request::Request(
+LayerTilesRequest::LayerTilesRequest(
     std::string mapId,
     std::string layerId,
     std::vector<TileId> tiles,
@@ -21,35 +21,51 @@ Request::Request(
       tiles_(std::move(tiles)),
       onResult_(std::move(onResult))
 {
+    if (tiles_.empty()) {
+        // An empty request is always set to success, but the client/service
+        // is responsible for triggering notifyStatus() in that case.
+        status_ = RequestStatus::Success;
+    }
 }
 
-bool Request::isDone() const
-{
-    return resultCount_ == tiles_.size();
-}
-
-void Request::notifyResult(TileFeatureLayer::Ptr r) {
+void LayerTilesRequest::notifyResult(TileFeatureLayer::Ptr r) {
     if (onResult_)
         onResult_(std::move(r));
     ++resultCount_;
-    if (isDone())
-        notifyDone();
+    if (resultCount_ == tiles_.size()) {
+        setStatus(RequestStatus::Success);
+    }
 }
 
-void Request::notifyDone()
+void LayerTilesRequest::setStatus(RequestStatus s)
 {
-    if (onDone_)
-        onDone_();
-    doneConditionVariable_.notify_all();
+    {
+        std::unique_lock statusLock(statusMutex_);
+        this->status_ = s;
+    }
+    notifyStatus();
 }
 
-void Request::wait()
+void LayerTilesRequest::notifyStatus()
 {
-    std::unique_lock doneLock(doneMutex_);
-    doneConditionVariable_.wait(doneLock, [this]{return isDone();});
+    if (isDone() && onDone_) {
+        // Run the final callback function.
+        onDone_(this->status_);
+    }
+    statusConditionVariable_.notify_all();
 }
 
-nlohmann::json Request::toJson()
+void LayerTilesRequest::wait()
+{
+    std::unique_lock doneLock(statusMutex_);
+    // Extra doneness check is to avoid infinite locking, e.g.
+    // because empty requests were not considered by calling method.
+    if (!isDone()) {
+        statusConditionVariable_.wait(doneLock, [this]{ return isDone(); });
+    }
+}
+
+nlohmann::json LayerTilesRequest::toJson()
 {
     auto tileIds = nlohmann::json::array();
     for (auto const& tid : tiles_)
@@ -61,13 +77,23 @@ nlohmann::json Request::toJson()
     });
 }
 
+RequestStatus LayerTilesRequest::getStatus()
+{
+    return this->status_;
+}
+
+bool LayerTilesRequest::isDone()
+{
+    return status_ != RequestStatus::Open;
+}
+
 struct Service::Controller
 {
-    using Job = std::pair<MapTileKey, Request::Ptr>;
+    using Job = std::pair<MapTileKey, LayerTilesRequest::Ptr>;
 
     std::set<MapTileKey> jobsInProgress_;    // Set of jobs currently in progress
     Cache::Ptr cache_;                       // The cache for the service
-    std::list<Request::Ptr> requests_;       // List of requests currently being processed
+    std::list<LayerTilesRequest::Ptr> requests_;       // List of requests currently being processed
     std::condition_variable jobsAvailable_;  // Condition variable to signal job availability
     std::mutex jobsMutex_;  // Mutex used with the jobsAvailable_ condition variable
 
@@ -79,12 +105,13 @@ struct Service::Controller
 
     std::optional<Job> nextJob(DataSourceInfo const& i)
     {
+        // Workers call the nextJob function when they are free.
         // Note: For thread safety, jobsMutex_ must be held
         //  when calling this function.
 
         std::optional<Job> result;
 
-        // Return next job if available
+        // Return next job, if available.
         bool cachedTilesServed = false;
         do {
             cachedTilesServed = false;
@@ -92,10 +119,13 @@ struct Service::Controller
                 auto& request = *reqIt;
                 auto layerIt = i.layers_.find(request->layerId_);
 
+                // Are there tiles left to be processed in the request?
                 if (request->mapId_ == i.mapId_ && layerIt != i.layers_.end()) {
-                    if (request->nextTileIndex_ >= request->tiles_.size())
+                    if (request->nextTileIndex_ >= request->tiles_.size()) {
                         continue;
+                    }
 
+                    // Create result wrapper object.
                     auto tileId = request->tiles_[request->nextTileIndex_++];
                     result = {MapTileKey(), request};
                     result->first.layer_ = layerIt->second->type_;
@@ -103,10 +133,10 @@ struct Service::Controller
                     result->first.layerId_ = request->layerId_;
                     result->first.tileId_ = tileId;
 
-                    // Cache lookup
+                    // Cache lookup.
                     auto cachedResult = cache_->getTileFeatureLayer(result->first, i);
                     if (cachedResult) {
-                        // TODO: Consider TTL
+                        // TODO: Consider TTL.
                         log().debug("Serving cached tile: {}", result->first.toString());
                         request->notifyResult(cachedResult);
                         result.reset();
@@ -115,11 +145,11 @@ struct Service::Controller
                     }
 
                     if (jobsInProgress_.find(result->first) != jobsInProgress_.end()) {
-                        // Don't work on something that is already being worked on. Instead,
-                        // wait for the work to finish, then send the (hopefully cached) result.
+                        // Don't work on something that is already being worked on.
+                        // Wait for the work to finish, then send the (hopefully cached) result.
                         log().debug("Delaying tile with job in progress: {}",
                                      result->first.toString());
-                        --request->nextTileIndex_;
+                        --(request->nextTileIndex_);
                         result.reset();
                         continue;
                     }
@@ -175,12 +205,13 @@ struct Service::Worker
                 lock,
                 [&, this]()
                 {
-                    if (shouldTerminate_)
+                    if (shouldTerminate_) {
+                        // Set by the controller at shutdown or if a data source
+                        // is removed. All worker instances are expected to terminate.
                         return true;
+                    }
                     nextJob = controller_.nextJob(info_);
-                    if (nextJob)
-                        return true;
-                    return false;
+                    return nextJob.has_value();
                 });
         }
 
@@ -200,9 +231,8 @@ struct Service::Worker
                 controller_.cache_->putTileFeatureLayer(result);
                 controller_.jobsInProgress_.erase(mapTileKey);
                 request->notifyResult(result);
-                // As we have now entered a tile into the cache, it
-                // is time to notify other workers that this tile can
-                // now be served.
+                // As we entered a tile into the cache, notify other workers
+                // that this tile can be served.
                 controller_.jobsAvailable_.notify_all();
             }
         }
@@ -218,8 +248,8 @@ struct Service::Worker
 
 struct Service::Impl : public Service::Controller
 {
-    std::map<DataSource::Ptr, DataSourceInfo> dataSourceInfo_;  // Map of data sources and their corresponding info
-    std::map<DataSource::Ptr, std::vector<Worker::Ptr>> dataSourceWorkers_; // Map of data sources and their corresponding workers
+    std::map<DataSource::Ptr, DataSourceInfo> dataSourceInfo_;
+    std::map<DataSource::Ptr, std::vector<Worker::Ptr>> dataSourceWorkers_;
 
     explicit Impl(Cache::Ptr cache) : Controller(std::move(cache)) {}
 
@@ -230,7 +260,8 @@ struct Service::Impl : public Service::Controller
                 worker->shouldTerminate_ = true;
             }
         }
-        jobsAvailable_.notify_all();  // Wake up all workers to check shouldTerminate_
+        // Wake up all workers to check shouldTerminate_.
+        jobsAvailable_.notify_all();
 
         for (auto& dataSourceAndWorkers : dataSourceWorkers_) {
             for (auto& worker : dataSourceAndWorkers.second) {
@@ -263,31 +294,32 @@ struct Service::Impl : public Service::Controller
 
         if (workers != dataSourceWorkers_.end())
         {
-            // Signal each worker thread to terminate
+            // Signal each worker thread to terminate.
             for (auto& worker : workers->second) {
                 worker->shouldTerminate_ = true;
             }
             jobsAvailable_.notify_all();
 
-            // Wait for each worker thread to terminate
+            // Wait for each worker thread to terminate.
             for (auto& worker : workers->second) {
                 if (worker->thread_.joinable()) {
                     worker->thread_.join();
                 }
             }
 
-            // Remove workers
+            // Remove workers.
             dataSourceWorkers_.erase(workers);
         }
     }
 
-    void addRequest(Request::Ptr r)
+    // All requests must be validated with canProcess before adding them!
+    void addRequest(LayerTilesRequest::Ptr r)
     {
         if (!r)
-            throw logRuntimeError("Attempt to call Service::addRequestFromJson(nullptr).");
+            throw logRuntimeError("Attempt to call Service::addRequest(nullptr).");
         if (r->isDone()) {
-            // Nothing to do
-            r->notifyDone();
+            // Nothing to do.
+            r->notifyStatus();
             return;
         }
 
@@ -298,10 +330,10 @@ struct Service::Impl : public Service::Controller
         jobsAvailable_.notify_all();
     }
 
-    void abortRequest(Request::Ptr const& r)
+    void abortRequest(LayerTilesRequest::Ptr const& r)
     {
         std::unique_lock lock(jobsMutex_);
-        // Simply Remove the request from the list of requests
+        // Remove the request from the list of requests.
         requests_.remove_if([r](auto&& request) { return r == request; });
     }
 
@@ -330,13 +362,30 @@ void Service::remove(const DataSource::Ptr& dataSource)
     impl_->removeDataSource(dataSource);
 }
 
-Request::Ptr Service::request(Request::Ptr r)
+bool Service::request(std::vector<LayerTilesRequest::Ptr> requests)
 {
-    impl_->addRequest(r);
-    return r;
+    bool dataSourcesAvailable = true;
+    for (const auto& r : requests) {
+        if (!hasLayer(r->mapId_, r->layerId_)) {
+            dataSourcesAvailable = false;
+            r->setStatus(RequestStatus::NoDataSource);
+        }
+    }
+    // Second pass either aborts requests or add all to job queue.
+    for (const auto& r : requests) {
+        if (!dataSourcesAvailable) {
+            if (r->getStatus() != RequestStatus::NoDataSource){
+                r->setStatus(RequestStatus::Aborted);
+            }
+        }
+        else {
+            impl_->addRequest(r);
+        }
+    }
+    return dataSourcesAvailable;
 }
 
-void Service::abort(const Request::Ptr& r)
+void Service::abort(const LayerTilesRequest::Ptr& r)
 {
     impl_->abortRequest(r);
 }
@@ -351,4 +400,19 @@ Cache::Ptr Service::cache()
     return impl_->cache_;
 }
 
+bool Service::hasLayer(std::string const& mapId, std::string const& layerId)
+{
+    std::unique_lock lock(impl_->jobsMutex_);
+    // Check that one of the data sources can fulfill the request.
+    for (auto& dataSourceAndInfo : impl_->dataSourceInfo_) {
+        auto& info = dataSourceAndInfo.second;
+        if (mapId != info.mapId_)
+            continue;
+        if (info.layers_.find(layerId) != info.layers_.end()) {
+            return true;
+        }
+    }
+    return false;
+}
+
 }  // namespace mapget


Test patch:
diff --git a/test/integration/CMakeLists.txt b/test/integration/CMakeLists.txt
index 8937a31f..d7533831 100644
--- a/test/integration/CMakeLists.txt
+++ b/test/integration/CMakeLists.txt
@@ -100,3 +100,5 @@ add_wheel_test(test-config-py
     # Request from cpp datasource
     -f "python -m mapget --config ${CMAKE_CURRENT_LIST_DIR}/../../examples/config/sample-first-datasource.toml fetch"
 )
+
+# TODO should there be a Python test for the 400 response?
diff --git a/test/unit/test-http-datasource.cpp b/test/unit/test-http-datasource.cpp
index 4bb674fa..edc96ea2 100644
--- a/test/unit/test-http-datasource.cpp
+++ b/test/unit/test-http-datasource.cpp
@@ -2,20 +2,18 @@
 #include "httplib.h"
 #include "mapget/log.h"
 
-#include "mapget/http-datasource/datasource-server.h"
 #include "mapget/http-datasource/datasource-client.h"
-#include "mapget/http-service/http-service.h"
+#include "mapget/http-datasource/datasource-server.h"
 #include "mapget/http-service/http-client.h"
+#include "mapget/http-service/http-service.h"
 #include "mapget/model/stream.h"
 
-using namespace mapget;
-
 TEST_CASE("HttpDataSource", "[HttpDataSource]")
 {
-    mapget::setLogLevel("trace", log());
+    mapget::setLogLevel("trace", mapget::log());
 
     // Create DataSourceInfo.
-    auto info = DataSourceInfo::fromJson(R"(
+    auto info = mapget::DataSourceInfo::fromJson(R"(
     {
         "mapId": "Tropico",
         "layers": {
@@ -47,15 +45,17 @@ TEST_CASE("HttpDataSource", "[HttpDataSource]")
     )"_json);
 
     // Initialize a DataSource.
-    DataSourceServer ds(info);
+    mapget::DataSourceServer ds(info);
     auto dataSourceRequestCount = 0;
-    ds.onTileRequest([&](auto&& tile) {
-        auto f = tile->newFeature("Way", {{"areaId", "Area42"}, {"wayId", 0}});
-        auto g = f->geom()->newGeometry(GeomType::Line);
-        g->append({42., 11});
-        g->append({42., 12});
-        ++dataSourceRequestCount;
-    });
+    ds.onTileRequest(
+        [&](auto&& tile)
+        {
+            auto f = tile->newFeature("Way", {{"areaId", "Area42"}, {"wayId", 0}});
+            auto g = f->geom()->newGeometry(mapget::GeomType::Line);
+            g->append({42., 11});
+            g->append({42., 12});
+            ++dataSourceRequestCount;
+        });
 
     // Launch the DataSource on a separate thread.
     ds.go();
@@ -70,7 +70,8 @@ TEST_CASE("HttpDataSource", "[HttpDataSource]")
 
         // Send a GET info request.
         auto fetchedInfoJson = cli.Get("/info");
-        auto fetchedInfo = DataSourceInfo::fromJson(nlohmann::json::parse(fetchedInfoJson->body));
+        auto fetchedInfo =
+            mapget::DataSourceInfo::fromJson(nlohmann::json::parse(fetchedInfoJson->body));
         REQUIRE(fetchedInfo.toJson() == info.toJson());
     }
 
@@ -88,7 +89,7 @@ TEST_CASE("HttpDataSource", "[HttpDataSource]")
 
         // Check the response body for expected content.
         auto receivedTileCount = 0;
-        TileLayerStream::Reader reader(
+        mapget::TileLayerStream::Reader reader(
             [&](auto&& mapId, auto&& layerId)
             {
                 REQUIRE(mapId == info.mapId_);
@@ -100,38 +101,66 @@ TEST_CASE("HttpDataSource", "[HttpDataSource]")
         REQUIRE(receivedTileCount == 1);
     }
 
-    SECTION("Stop the data source")
+    SECTION("Query mapget HTTP service")
     {
-        ds.stop();
-        REQUIRE(ds.isRunning() == false);
+        mapget::HttpService service;
+        service.add(std::make_shared<mapget::RemoteDataSource>("localhost", ds.port()));
+
+        service.go();
+
+        SECTION("Query through mapget HTTP service")
+        {
+            mapget::HttpClient client("localhost", service.port());
+
+            auto receivedTileCount = 0;
+            client.request(std::make_shared<mapget::LayerTilesRequest>(
+                               "Tropico",
+                               "WayLayer",
+                               std::vector<mapget::TileId>{{1234, 5678, 9112, 1234}},
+                               [&](auto&& tile) { receivedTileCount++; }))
+                ->wait();
+
+            REQUIRE(receivedTileCount == 4);
+            // One tile requested twice, so the cache was used.
+            REQUIRE(dataSourceRequestCount == 3);
+        }
+
+        SECTION("Trigger 400 responses")
+        {
+            mapget::HttpClient client("localhost", service.port());
+
+            auto receivedTileCount = 0;
+            auto unknownMapRequest = std::make_shared<mapget::LayerTilesRequest>(
+                "UnknownMap",
+                "WayLayer",
+                std::vector<mapget::TileId>{{1234}},
+                [&](auto&& tile) { receivedTileCount++; });
+            client.request(unknownMapRequest)->wait();
+            REQUIRE(unknownMapRequest->getStatus() == mapget::RequestStatus::NoDataSource);
+            REQUIRE(receivedTileCount == 0);
+
+            auto unknownLayerRequest = std::make_shared<mapget::LayerTilesRequest>(
+                "Tropico",
+                "UnknownLayer",
+                std::vector<mapget::TileId>{{1234}},
+                [&](auto&& tile) { receivedTileCount++; });
+            client.request(unknownLayerRequest)->wait();
+            REQUIRE(unknownLayerRequest->getStatus() == mapget::RequestStatus::NoDataSource);
+            REQUIRE(receivedTileCount == 0);
+        }
+
+        service.stop();
+        REQUIRE(service.isRunning() == false);
     }
 
     SECTION("Wait for data source")
     {
-        auto waitThread = std::thread([&]{ds.waitForSignal();});
+        auto waitThread = std::thread([&] { ds.waitForSignal(); });
         ds.stop();
         waitThread.join();
         REQUIRE(ds.isRunning() == false);
     }
 
-    SECTION("Contact through HTTP service")
-    {
-        HttpService service;
-        service.add(std::make_shared<RemoteDataSource>("localhost", ds.port()));
-        service.go();
-
-        HttpClient client("localhost", service.port());
-
-        auto receivedTileCount = 0;
-        client.request(std::make_shared<Request>(
-            "Tropico",
-            "WayLayer",
-            std::vector<TileId>{{1234, 5678, 9112, 1234}},
-            [&](auto&& tile) { receivedTileCount++; }
-        ))->wait();
-
-        REQUIRE(receivedTileCount == 4);
-        REQUIRE(dataSourceRequestCount == 3); // One tile requested twice, so the cache was used.
-        service.stop();
-    }
+    ds.stop();
+    REQUIRE(ds.isRunning() == false);
 }