rest: allow reading partial block data from storage

diff --git a/doc/REST-interface.md b/doc/REST-interface.md
index 0fee7812..ed46e222 100644
--- a/doc/REST-interface.md
+++ b/doc/REST-interface.md
@@ -47,6 +47,11 @@ The HTTP request and response are both handled entirely in-memory.
 
 With the /notxdetails/ option JSON response will only contain the transaction hash instead of the complete transaction details. The option only affects the JSON response.
 
+- `GET /rest/blockpart/<BLOCK-HASH>.<bin|hex>?offset=<OFFSET>&size=<SIZE>`
+
+Given a block hash: returns a block part, in binary or hex-encoded binary formats.
+Responds with 404 if the block or the byte range doesn't exist.
+
 #### Blockheaders
 `GET /rest/headers/<BLOCK-HASH>.<bin|hex|json>?count=<COUNT=5>`
 
diff --git a/doc/release-notes-33657.md b/doc/release-notes-33657.md
new file mode 100644
index 00000000..f9e6841b
--- /dev/null
+++ b/doc/release-notes-33657.md
@@ -0,0 +1,5 @@
+New REST API
+------------
+
+- A new REST API endpoint (`/rest/blockpart/BLOCKHASH.bin?offset=X&size=Y`) has been introduced
+  for efficiently fetching a range of bytes from block `BLOCKHASH`.
diff --git a/src/rest.cpp b/src/rest.cpp
index 26312633..06da3906 100644
--- a/src/rest.cpp
+++ b/src/rest.cpp
@@ -379,10 +379,17 @@ static bool rest_spent_txouts(const std::any& context, HTTPRequest* req, const s
     }
 }
 
+/**
+ * This handler is used by multiple HTTP endpoints:
+ * - `/block/` via `rest_block_extended()`
+ * - `/block/notxdetails/` via `rest_block_notxdetails()`
+ * - `/blockpart/` via `rest_block_part()` (doesn't support JSON response, so `tx_verbosity` is unset)
+ */
 static bool rest_block(const std::any& context,
                        HTTPRequest* req,
                        const std::string& uri_part,
-                       TxVerbosity tx_verbosity)
+                       std::optional<TxVerbosity> tx_verbosity,
+                       std::optional<std::pair<size_t, size_t>> block_part = std::nullopt)
 {
     if (!CheckWarmup(req))
         return false;
@@ -416,12 +423,14 @@ static bool rest_block(const std::any& context,
         pos = pblockindex->GetBlockPos();
     }
 
-    const auto block_data{chainman.m_blockman.ReadRawBlock(pos)};
+    const auto block_data{chainman.m_blockman.ReadRawBlock(pos, block_part)};
     if (!block_data) {
         switch (block_data.error()) {
         case node::ReadRawError::IO: return RESTERR(req, HTTP_INTERNAL_SERVER_ERROR, "I/O error reading " + hashStr);
-        case node::ReadRawError::BadPartRange: break; // can happen only when reading a block part
-        }
+        case node::ReadRawError::BadPartRange:
+            assert(block_part);
+            return RESTERR(req, HTTP_BAD_REQUEST, strprintf("Bad block part offset/size %d/%d for %s", block_part->first, block_part->second, hashStr));
+        } // no default case, so the compiler can warn about missing cases
         assert(false);
     }
 
@@ -440,14 +449,17 @@ static bool rest_block(const std::any& context,
     }
 
     case RESTResponseFormat::JSON: {
-        CBlock block{};
-        DataStream block_stream{*block_data};
-        block_stream >> TX_WITH_WITNESS(block);
-        UniValue objBlock = blockToJSON(chainman.m_blockman, block, *tip, *pblockindex, tx_verbosity, chainman.GetConsensus().powLimit);
-        std::string strJSON = objBlock.write() + "\n";
-        req->WriteHeader("Content-Type", "application/json");
-        req->WriteReply(HTTP_OK, strJSON);
-        return true;
+        if (tx_verbosity) {
+            CBlock block{};
+            DataStream block_stream{*block_data};
+            block_stream >> TX_WITH_WITNESS(block);
+            UniValue objBlock = blockToJSON(chainman.m_blockman, block, *tip, *pblockindex, *tx_verbosity, chainman.GetConsensus().powLimit);
+            std::string strJSON = objBlock.write() + "\n";
+            req->WriteHeader("Content-Type", "application/json");
+            req->WriteReply(HTTP_OK, strJSON);
+            return true;
+        }
+        return RESTERR(req, HTTP_BAD_REQUEST, "JSON output is not supported for this request type");
     }
 
     default: {
@@ -466,6 +478,25 @@ static bool rest_block_notxdetails(const std::any& context, HTTPRequest* req, co
     return rest_block(context, req, uri_part, TxVerbosity::SHOW_TXID);
 }
 
+static bool rest_block_part(const std::any& context, HTTPRequest* req, const std::string& uri_part)
+{
+    try {
+        if (const auto opt_offset{ToIntegral<size_t>(req->GetQueryParameter("offset").value_or(""))}) {
+            if (const auto opt_size{ToIntegral<size_t>(req->GetQueryParameter("size").value_or(""))}) {
+                return rest_block(context, req, uri_part,
+                                  /*tx_verbosity=*/std::nullopt,
+                                  /*block_part=*/{{*opt_offset, *opt_size}});
+            } else {
+                return RESTERR(req, HTTP_BAD_REQUEST, "Block part size missing or invalid");
+            }
+        } else {
+            return RESTERR(req, HTTP_BAD_REQUEST, "Block part offset missing or invalid");
+        }
+    } catch (const std::runtime_error& e) {
+        return RESTERR(req, HTTP_BAD_REQUEST, e.what());
+    }
+}
+
 static bool rest_filter_header(const std::any& context, HTTPRequest* req, const std::string& uri_part)
 {
     if (!CheckWarmup(req)) return false;
@@ -1114,6 +1145,7 @@ static const struct {
       {"/rest/tx/", rest_tx},
       {"/rest/block/notxdetails/", rest_block_notxdetails},
       {"/rest/block/", rest_block_extended},
+      {"/rest/blockpart/", rest_block_part},
       {"/rest/blockfilter/", rest_block_filter},
       {"/rest/blockfilterheaders/", rest_filter_header},
       {"/rest/chaininfo", rest_chaininfo},
diff --git a/test/functional/interface_rest.py b/test/functional/interface_rest.py
index 54fc908a..be5bb78b 100755
--- a/test/functional/interface_rest.py
+++ b/test/functional/interface_rest.py
@@ -28,7 +28,6 @@ from test_framework.wallet import (
     MiniWallet,
     getnewdestination,
 )
-from typing import Optional
 
 
 INVALID_PARAM = "abc"
@@ -66,13 +65,16 @@ class RESTTest (BitcoinTestFramework):
             body: str = '',
             status: int = 200,
             ret_type: RetType = RetType.JSON,
-            query_params: Optional[dict[str, typing.Any]] = None,
+            query_params: typing.Union[dict[str, typing.Any], str, None] = None,
             ) -> typing.Union[http.client.HTTPResponse, bytes, str, None]:
         rest_uri = '/rest' + uri
         if req_type in ReqType:
             rest_uri += f'.{req_type.name.lower()}'
         if query_params:
-            rest_uri += f'?{urllib.parse.urlencode(query_params)}'
+            if isinstance(query_params, str):
+                rest_uri += f'?{query_params}'
+            else:
+                rest_uri += f'?{urllib.parse.urlencode(query_params)}'
 
         conn = http.client.HTTPConnection(self.url.hostname, self.url.port)
         self.log.debug(f'{http_method} {rest_uri} {body}')
@@ -82,7 +84,7 @@ class RESTTest (BitcoinTestFramework):
             conn.request('POST', rest_uri, body)
         resp = conn.getresponse()
 
-        assert_equal(resp.status, status)
+        assert resp.status == status, f"Expected: {status}, Got: {resp.status} - Response: {str(resp.read())}"
 
         if ret_type == RetType.OBJ:
             return resp
@@ -455,6 +457,52 @@ class RESTTest (BitcoinTestFramework):
                 expected = [(p["scriptPubKey"], p["value"]) for p in prevouts]
                 assert_equal(expected, actual)
 
+        self.log.info("Test the /blockpart URI")
+
+        blockhash = self.nodes[0].getbestblockhash()
+        block_bin = self.test_rest_request(f"/block/{blockhash}", req_type=ReqType.BIN, ret_type=RetType.BYTES)
+        for req_type in (ReqType.BIN, ReqType.HEX):
+            def get_block_part(status: int = 200, **kwargs):
+                resp = self.test_rest_request(f"/blockpart/{blockhash}", status=status,
+                                              req_type=req_type, ret_type=RetType.BYTES, **kwargs)
+                assert isinstance(resp, bytes)
+                if req_type is ReqType.HEX and status == 200:
+                    resp = bytes.fromhex(resp.decode().strip())
+                return resp
+
+            assert_equal(block_bin, get_block_part(query_params={"offset": 0, "size": len(block_bin)}))
+
+            assert len(block_bin) >= 500
+            assert_equal(block_bin[20:320], get_block_part(query_params={"offset": 20, "size": 300}))
+            assert_equal(block_bin[-5:], get_block_part(query_params={"offset": len(block_bin) - 5, "size": 5}))
+
+            get_block_part(status=400, query_params={"offset": 10})
+            get_block_part(status=400, query_params={"size": 100})
+            get_block_part(status=400, query_params={"offset": "x"})
+            get_block_part(status=400, query_params={"size": "y"})
+            get_block_part(status=400, query_params={"offset": "x", "size": "y"})
+            assert get_block_part(status=400, query_params="%XY").decode("utf-8").startswith("URI parsing failed")
+
+            get_block_part(status=400, query_params={"offset": 0, "size": 0})
+            get_block_part(status=400, query_params={"offset": len(block_bin), "size": 0})
+            get_block_part(status=400, query_params={"offset": len(block_bin) + 1, "size": 1})
+            get_block_part(status=400, query_params={"offset": len(block_bin), "size": 1})
+            get_block_part(status=400, query_params={"offset": len(block_bin) + 1, "size": 1})
+            get_block_part(status=400, query_params={"offset": 0, "size": len(block_bin) + 1})
+
+        self.test_rest_request(f"/blockpart/{blockhash}", status=400, req_type=ReqType.JSON, ret_type=RetType.OBJ)
+
+        self.log.info("Missing block data should cause REST API to fail")
+
+        self.test_rest_request(f"/block/{blockhash}", status=200, req_type=ReqType.BIN, ret_type=RetType.OBJ)
+        self.test_rest_request(f"/blockpart/{blockhash}", query_params={"offset": 0, "size": 1}, status=200, req_type=ReqType.BIN, ret_type=RetType.OBJ)
+        blk_files = list(self.nodes[0].blocks_path.glob("blk*.dat"))
+        for blk_file in blk_files:
+            blk_file.rename(blk_file.with_suffix('.bkp'))
+        self.test_rest_request(f"/block/{blockhash}", status=500, req_type=ReqType.BIN, ret_type=RetType.OBJ)
+        self.test_rest_request(f"/blockpart/{blockhash}", query_params={"offset": 0, "size": 1}, status=500, req_type=ReqType.BIN, ret_type=RetType.OBJ)
+        for blk_file in blk_files:
+            blk_file.with_suffix('.bkp').rename(blk_file)
 
         self.log.info("Test the /deploymentinfo URI")