diff --git a/CHANGELOG.md b/CHANGELOG.md index 126fdff584c8..e7699e17e9b4 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -16,6 +16,7 @@ and this project adheres to [Semantic Versioning](http://semver.org/spec/v2.0.0. ### Added - C++ endpoints can now use `ccf::endpoints::Endpoint::add_openapi_response()` to document additional HTTP responses in their generated OpenAPI schema without changing the endpoint's primary success response (#8115). +- Ledger chunk download clients can opt in to immutable `.committed_prefix` resources containing recent committed entries that are not yet available in canonical `.committed` files. (#8214) - New `ledger.max_transaction_size` node configuration option (default `32MB`), which caps the total serialised size of transactions written to the ledger. The limit covers the whole ledger entry: the fixed 8-byte ledger entry header, the ledger encryption header, public domain size field, public domain and encrypted private domain. It is checked before a transaction is applied, so an oversized transaction is now rejected with `413 Payload Too Large` and error code `TransactionTooLarge`, and subsequent transactions are unaffected, where previously an excessively large transaction could terminate the node. Reserved internal signature transactions are exempt because they must fill their reserved ledger version. The limit applies only to newly serialised non-reserved transactions; deserialising existing entries (including during recovery), historical queries and snapshots are unaffected, so entries written under a larger or unset limit remain readable. It must be smaller than `memory.max_msg_size` by at least the ring-buffer range response overhead, which is validated at node startup and by `--check` (#7992). ### Changed diff --git a/doc/operations/configuration.rst b/doc/operations/configuration.rst index fb0b0d1bf00a..4dffb943064c 100644 --- a/doc/operations/configuration.rst +++ b/doc/operations/configuration.rst @@ -71,7 +71,7 @@ The `enabled_operator_features` configuration field allows enabling or disabling Currently supported features are: 1. 'SnapshotRead': gates access to endpoints used to fetch snapshots directly from nodes (:http:GET:`/node/snapshot`, :http:HEAD:`/node/snapshot`, :http:GET:`/node/snapshot/{snapshot_name}` and :http:HEAD:`/node/snapshot/{snapshot_name}`). -2. 'LedgerChunkRead': gates access to endpoints used to retrieve ledger chunks (:http:GET:`/node/ledger_chunk`, :http:HEAD:`/node/ledger_chunk`, :http:GET:`/node/ledger_chunk/{chunk_name}` and :http:HEAD:`/node/ledger_chunk/{chunk_name}`). +2. 'LedgerChunkRead': gates access to endpoints used to retrieve ledger chunks (:http:GET:`/node/ledger_chunk`, :http:HEAD:`/node/ledger_chunk`, :http:GET:`/node/ledger_chunk/{chunk_name}`, :http:HEAD:`/node/ledger_chunk/{chunk_name}`, :http:GET:`/node/ledger_chunk/committed_prefix/{chunk_name}` and :http:HEAD:`/node/ledger_chunk/committed_prefix/{chunk_name}`). 3. 'SnapshotCreate': gates access to the operator endpoint used to create a snapshot on the next signature transaction (:http:POST:`/node/snapshot:create`). Since these operations may require disk IO and produce large responses, these features should not be enabled on interfaces with public access, and instead restricted to interfaces with local connectivity for node-to-node and operator access. @@ -100,4 +100,3 @@ A rolling upgrade from ``Dual`` to ``CoseOnly`` is a two-step process: 1. **CoseAllowDualJoin.** Deploy a binary that returns ``CoseAllowDualJoin``. Replace nodes one at a time. During this phase, new nodes running the old ``Dual`` binary can still join as replacements. 2. **CoseOnly.** Once all nodes are upgraded, deploy a binary that returns ``CoseOnly``. Replace nodes again. After this, ``Dual`` joiners are rejected. - diff --git a/doc/operations/ledger_snapshot.rst b/doc/operations/ledger_snapshot.rst index 15c58e6acf54..8aacbe3e8d88 100644 --- a/doc/operations/ledger_snapshot.rst +++ b/doc/operations/ledger_snapshot.rst @@ -114,10 +114,33 @@ These endpoints allow downloading a specific ledger chunk by name, where ``_). +The canonical and committed-prefix chunk endpoints also support the ``Want-Repr-Digest`` request header (`RFC 9530 `_). When set, the response will include a ``Repr-Digest`` header containing the digest of the full representation of the file. Supported algorithms are ``sha-256``, ``sha-384``, and ``sha-512``. If the header contains only unsupported or invalid algorithms, the server defaults to ``sha-256`` (as permitted by `RFC 9530 Appendix C.2 `_). For example, a client sending ``Want-Repr-Digest: sha-256=1`` will receive a header such as ``Repr-Digest: sha-256=:AEGPTgUMw5e96wxZuDtpfm23RBU3nFwtgY5fw4NYORo=:`` in the response. diff --git a/doc/schemas/node_openapi.json b/doc/schemas/node_openapi.json index 0bc4626534ef..4f80ad9a40c3 100644 --- a/doc/schemas/node_openapi.json +++ b/doc/schemas/node_openapi.json @@ -919,7 +919,7 @@ "info": { "description": "This API provides public, uncredentialed access to service and node state.", "title": "CCF Public Node API", - "version": "5.0.6" + "version": "5.0.7" }, "openapi": "3.0.0", "paths": { @@ -1192,7 +1192,7 @@ }, "/node/ledger_chunk": { "get": { - "description": "Redirect to the corresponding /node/ledger_chunk/{chunk_name} endpoint for the ledger chunk including the sequence number specified in the 'since' query parameter.", + "description": "Redirect to the corresponding /node/ledger_chunk/{chunk_name} endpoint for the ledger chunk including the sequence number specified in the 'since' query parameter. If 'include_committed_prefix' is true and no committed file is available, this may temporarily redirect to a synthetic committed-prefix resource.", "operationId": "GetNodeLedgerChunk", "parameters": [ { @@ -1202,9 +1202,20 @@ "schema": { "$ref": "#/components/schemas/uint64" } + }, + { + "in": "query", + "name": "include_committed_prefix", + "required": false, + "schema": { + "$ref": "#/components/schemas/boolean" + } } ], "responses": { + "307": { + "description": "Redirect to a temporary committed ledger prefix." + }, "308": { "description": "Redirect to the selected ledger chunk." }, @@ -1221,7 +1232,7 @@ } }, "head": { - "description": "Redirect to the corresponding /node/ledger_chunk/{chunk_name} endpoint for the ledger chunk including the sequence number specified in the 'since' query parameter.", + "description": "Redirect to the corresponding /node/ledger_chunk/{chunk_name} endpoint for the ledger chunk including the sequence number specified in the 'since' query parameter. If 'include_committed_prefix' is true and no committed file is available, this may temporarily redirect to a synthetic committed-prefix resource.", "operationId": "HeadNodeLedgerChunk", "parameters": [ { @@ -1231,9 +1242,20 @@ "schema": { "$ref": "#/components/schemas/uint64" } + }, + { + "in": "query", + "name": "include_committed_prefix", + "required": false, + "schema": { + "$ref": "#/components/schemas/boolean" + } } ], "responses": { + "307": { + "description": "Redirect to a temporary committed ledger prefix." + }, "308": { "description": "Redirect to the selected ledger chunk." }, @@ -1250,6 +1272,82 @@ } } }, + "/node/ledger_chunk/committed_prefix/{chunk_name}": { + "get": { + "description": "Download a synthetic chunk containing only committed ledger entries. Supports HTTP Range and digest headers. The resource is not a canonical .committed ledger file and must not be used for recovery.", + "operationId": "GetNodeLedgerChunkCommittedPrefixChunkName", + "responses": { + "200": { + "content": { + "application/octet-stream": { + "schema": { + "$ref": "#/components/schemas/Binary" + } + } + }, + "description": "The requested committed ledger prefix." + }, + "206": { + "content": { + "application/octet-stream": { + "schema": { + "$ref": "#/components/schemas/Binary" + } + } + }, + "description": "The requested byte range of the committed ledger prefix." + }, + "304": { + "description": "The requested committed ledger prefix has not changed." + }, + "404": { + "description": "The requested committed ledger prefix is not available." + }, + "default": { + "$ref": "#/components/responses/default" + } + }, + "summary": "Download committed ledger prefix", + "x-ccf-forwarding": { + "$ref": "#/components/x-ccf-forwarding/never" + } + }, + "head": { + "description": "Metadata about a synthetic chunk containing only committed ledger entries. The resource is not a canonical .committed ledger file.", + "operationId": "HeadNodeLedgerChunkCommittedPrefixChunkName", + "responses": { + "200": { + "description": "Metadata for the requested committed ledger prefix." + }, + "206": { + "description": "Metadata for the requested committed ledger prefix range." + }, + "304": { + "description": "The requested committed ledger prefix has not changed." + }, + "404": { + "description": "The requested committed ledger prefix is not available." + }, + "default": { + "$ref": "#/components/responses/default" + } + }, + "summary": "Committed ledger prefix metadata", + "x-ccf-forwarding": { + "$ref": "#/components/x-ccf-forwarding/never" + } + }, + "parameters": [ + { + "in": "path", + "name": "chunk_name", + "required": true, + "schema": { + "type": "string" + } + } + ] + }, "/node/ledger_chunk/{chunk_name}": { "get": { "description": "Download a specific ledger chunk by name. Supports HTTP Range header for partial downloads.", diff --git a/include/ccf/http_consts.h b/include/ccf/http_consts.h index c30ea896b917..df4de94d9043 100644 --- a/include/ccf/http_consts.h +++ b/include/ccf/http_consts.h @@ -33,6 +33,8 @@ namespace ccf static constexpr auto CCF_SNAPSHOT_NAME = "x-ms-ccf-snapshot-name"; static constexpr auto CCF_LEDGER_CHUNK_NAME = "x-ms-ccf-ledger-chunk-name"; + static constexpr auto CCF_LEDGER_CHUNK_KIND = + "x-ms-ccf-ledger-chunk-kind"; } namespace headervalues::contenttype diff --git a/python/src/ccf/ledger.py b/python/src/ccf/ledger.py index 8f63f8e70749..c5e5428e1d63 100644 --- a/python/src/ccf/ledger.py +++ b/python/src/ccf/ledger.py @@ -58,6 +58,7 @@ def __getattr__(name: str): SERVICE_INFO_TABLE_NAME = "public:ccf.gov.service.info" COMMITTED_FILE_SUFFIX = ".committed" +COMMITTED_PREFIX_FILE_SUFFIX = ".committed_prefix" RECOVERY_FILE_SUFFIX = ".recovery" IGNORED_FILE_SUFFIX = ".ignored" @@ -157,13 +158,29 @@ def unpack_array(buf, fmt): def range_from_filename(filename: str) -> tuple[int, int | None]: - elements = ( - os.path.basename(filename) - .replace(COMMITTED_FILE_SUFFIX, "") - .replace(RECOVERY_FILE_SUFFIX, "") - .replace("ledger_", "") - .split("-") - ) + basename = os.path.basename(filename) + is_recovery = basename.endswith(RECOVERY_FILE_SUFFIX) + basename = basename.removesuffix(RECOVERY_FILE_SUFFIX) + if basename.endswith(COMMITTED_PREFIX_FILE_SUFFIX): + if is_recovery: + raise ValueError(f"Could not read seqno range from ledger file {filename}") + + range_str = basename.removesuffix(COMMITTED_PREFIX_FILE_SUFFIX) + if not range_str.startswith("ledger_"): + raise ValueError(f"Could not read seqno range from ledger file {filename}") + + elements = range_str[len("ledger_") :].split("-") + if ( + len(elements) != 2 + or not all(element.isascii() and element.isdigit() for element in elements) + or int(elements[0]) == 0 + or int(elements[1]) < int(elements[0]) + ): + raise ValueError(f"Could not read seqno range from ledger file {filename}") + return (int(elements[0]), int(elements[1])) + + basename = basename.removesuffix(COMMITTED_FILE_SUFFIX) + elements = basename.replace("ledger_", "").split("-") if len(elements) == 2: return (int(elements[0]), int(elements[1])) elif len(elements) == 1: diff --git a/python/tests/test_ledger.py b/python/tests/test_ledger.py new file mode 100644 index 000000000000..a3753934deaf --- /dev/null +++ b/python/tests/test_ledger.py @@ -0,0 +1,34 @@ +# Copyright (c) Microsoft Corporation. All rights reserved. +# Licensed under the Apache 2.0 License. + +"""Unit tests for CCF ledger filename handling.""" + +import ccf.ledger +import pytest + + +def test_committed_prefix_filename_range(): + """Committed-prefix names expose their closed sequence-number range.""" + assert ccf.ledger.range_from_filename("ledger_42-100.committed_prefix") == (42, 100) + + +def test_committed_prefix_is_not_canonical_committed_file(): + """Committed prefixes stay excluded from committed-only discovery.""" + assert not ccf.ledger.is_ledger_chunk_committed("ledger_42-100.committed_prefix") + + +@pytest.mark.parametrize( + "filename", + [ + "ledger_0-100.committed_prefix", + "ledger_42-41.committed_prefix", + "ledger_42.committed_prefix", + "ledger_ledger_42-100.committed_prefix", + "ledger_42x-100.committed_prefix", + "ledger_42-100.committed_prefix.recovery", + ], +) +def test_committed_prefix_filename_rejects_invalid_ranges(filename: str): + """Committed-prefix names must contain one valid, closed range.""" + with pytest.raises(ValueError): + ccf.ledger.range_from_filename(filename) diff --git a/src/host/ledger.h b/src/host/ledger.h index f2ddd72415c6..a49823d53db2 100644 --- a/src/host/ledger.h +++ b/src/host/ledger.h @@ -19,6 +19,7 @@ #include #include #include +#include #include #include #include @@ -45,6 +46,7 @@ namespace asynchost auto f_name = f.path().filename(); if ( is_ledger_file_name_ignored(f_name) || + is_ledger_file_name_committed_prefix(f_name) || (!allow_recovery_files && is_ledger_file_name_recovery(f_name))) { continue; @@ -534,6 +536,90 @@ namespace asynchost return LedgerReadResult{entries, to_}; } + std::optional> read_entries_as_completed_chunk( + size_t from, size_t to) + { + std::unique_lock guard(file_lock); + + const auto [raw_entries_size, end_idx] = entries_size(from, to); + if (raw_entries_size == 0 || end_idx != to) + { + return std::nullopt; + } + + const auto entry_count = to - from + 1; + if ( + raw_entries_size > + std::numeric_limits::max() - sizeof(positions_offset_header_t)) + { + throw std::logic_error(fmt::format( + "Ledger entry range {}-{} is too large to represent as a chunk", + from, + to)); + } + const auto positions_offset = + sizeof(positions_offset_header_t) + raw_entries_size; + if ( + positions_offset > std::numeric_limits::max() || + entry_count > (std::numeric_limits::max() - positions_offset) / + sizeof(positions.at(0))) + { + throw std::logic_error(fmt::format( + "Ledger entry range {}-{} is too large to represent as a chunk", + from, + to)); + } + + const auto positions_size = entry_count * sizeof(positions.at(0)); + std::vector chunk(positions_offset + positions_size); + auto* out = chunk.data(); + auto remaining = chunk.size(); + serialized::write(out, remaining, positions_offset); + + if (fseeko(file, positions.at(from - start_idx), SEEK_SET) != 0) + { + throw std::logic_error(fmt::format( + "Failed to seek to entry {} in ledger file {}", from, file_name)); + } + + { + TimeBoundLogger log_if_slow(fmt::format( + "Reading committed ledger prefix {} to {} ({} bytes) - fread({})", + from, + to, + raw_entries_size, + file_name)); + if (fread(out, raw_entries_size, 1, file) != 1) + { + throw std::logic_error(fmt::format( + "Failed to read entry range {}-{} from ledger file {}", + from, + to, + file_name)); + } + } + out += raw_entries_size; + remaining -= raw_entries_size; + + const auto first_position = positions.at(from - start_idx); + for (size_t idx = from; idx <= to; ++idx) + { + const auto relative_position = sizeof(positions_offset_header_t) + + positions.at(idx - start_idx) - first_position; + if (relative_position > std::numeric_limits::max()) + { + throw std::logic_error(fmt::format( + "Entry {} offset is too large to represent in a ledger chunk", + idx)); + } + + serialized::write( + out, remaining, static_cast(relative_position)); + } + + return chunk; + } + bool truncate(size_t idx, bool remove_file_if_empty = true) { if ( @@ -959,7 +1045,9 @@ namespace asynchost return idx >= f->get_start_idx(); }); - if (f != files.rend() && idx <= (*f)->get_last_idx()) + if ( + f != files.rend() && (*f)->get_start_idx() <= idx && + idx <= (*f)->get_last_idx()) { return *f; } @@ -1732,6 +1820,49 @@ namespace asynchost return ledger_dir / name.value(); } + [[nodiscard]] std::optional> + committed_ledger_prefix_range_with_idx(size_t idx) + { + std::unique_lock guard(state_lock); + + if (idx == 0 || idx <= end_of_committed_files_idx || idx > committed_idx) + { + return std::nullopt; + } + + const auto it = get_it_contains_idx(idx); + if ( + it == files.end() || (*it)->get_start_idx() > idx || + (*it)->is_committed() || (*it)->is_recovery()) + { + return std::nullopt; + } + + return std::make_pair( + idx, std::min(committed_idx, (*it)->get_last_idx())); + } + + [[nodiscard]] std::optional> + read_committed_ledger_prefix(size_t from, size_t to) + { + std::unique_lock guard(state_lock); + + if (from == 0 || to < from || to > committed_idx) + { + return std::nullopt; + } + + const auto file = get_file_from_idx(from); + if ( + file == nullptr || file->get_start_idx() > from || + file->get_last_idx() < to || file->is_recovery()) + { + return std::nullopt; + } + + return file->read_entries_as_completed_chunk(from, to); + } + [[nodiscard]] size_t get_init_idx() { std::unique_lock guard(state_lock); diff --git a/src/host/ledger_filenames.h b/src/host/ledger_filenames.h index d544ea78f600..3f8b5b661559 100644 --- a/src/host/ledger_filenames.h +++ b/src/host/ledger_filenames.h @@ -2,18 +2,21 @@ // Licensed under the Apache 2.0 License. #pragma once +#include #include #include #include #include #include #include +#include namespace asynchost { namespace fs = std::filesystem; static constexpr auto ledger_committed_suffix = ".committed"; + static constexpr auto ledger_committed_prefix_suffix = ".committed_prefix"; static constexpr auto ledger_start_idx_delimiter = "_"; static constexpr auto ledger_last_idx_delimiter = "-"; static constexpr auto ledger_recovery_file_suffix = ".recovery"; @@ -50,6 +53,59 @@ namespace asynchost return file_name.ends_with(ledger_committed_suffix); } + static inline bool is_ledger_file_name_committed_prefix( + const std::string& file_name) + { + return file_name.ends_with(ledger_committed_prefix_suffix); + } + + static inline std::optional> + get_ledger_committed_prefix_range_from_file_name(std::string_view file_name) + { + static constexpr std::string_view prefix = "ledger_"; + static constexpr std::string_view suffix = ledger_committed_prefix_suffix; + + if (!file_name.starts_with(prefix) || !file_name.ends_with(suffix)) + { + return std::nullopt; + } + + file_name.remove_prefix(prefix.size()); + file_name.remove_suffix(suffix.size()); + + const auto delimiter = file_name.find(ledger_last_idx_delimiter); + if ( + delimiter == std::string_view::npos || delimiter == 0 || + delimiter == file_name.size() - 1 || + file_name.find(ledger_last_idx_delimiter, delimiter + 1) != + std::string_view::npos) + { + return std::nullopt; + } + + const auto parse_idx = [](std::string_view value) -> std::optional { + size_t idx = 0; + const auto* const end = value.data() + value.size(); + const auto [ptr, ec] = std::from_chars(value.data(), end, idx); + if (ec != std::errc() || ptr != end) + { + return std::nullopt; + } + return idx; + }; + + const auto start_idx = parse_idx(file_name.substr(0, delimiter)); + const auto end_idx = parse_idx(file_name.substr(delimiter + 1)); + if ( + !start_idx.has_value() || !end_idx.has_value() || + start_idx.value() == 0 || end_idx.value() < start_idx.value()) + { + return std::nullopt; + } + + return std::make_pair(start_idx.value(), end_idx.value()); + } + static inline bool is_ledger_file_name_recovery(const std::string& file_name) { return file_name.ends_with(ledger_recovery_file_suffix); @@ -64,7 +120,8 @@ namespace asynchost { // Catch-all for all files that should be ignored return is_ledger_file_name_recovery(file_name) || - is_ledger_file_name_ignored(file_name); + is_ledger_file_name_ignored(file_name) || + is_ledger_file_name_committed_prefix(file_name); } static inline fs::path remove_suffix( diff --git a/src/host/test/files_cleanup_test.cpp b/src/host/test/files_cleanup_test.cpp index c1951b28630b..82e14cd02af7 100644 --- a/src/host/test/files_cleanup_test.cpp +++ b/src/host/test/files_cleanup_test.cpp @@ -700,6 +700,36 @@ TEST_CASE("is_ledger_file_name_committed: detects committed suffix") CHECK_FALSE(is_ledger_file_name_committed("ledger_1-100.committed.ignored")); } +TEST_CASE("committed prefix names contain a strict range") +{ + const auto range = get_ledger_committed_prefix_range_from_file_name( + "ledger_42-100.committed_prefix"); + REQUIRE(range.has_value()); + CHECK(range->first == 42); + CHECK(range->second == 100); + + for (const auto* invalid_name : + {"ledger_0-100.committed_prefix", + "ledger_42-41.committed_prefix", + "ledger_42.committed_prefix", + "ledger_42-100.committed", + "ledger_42-100-101.committed_prefix", + "ledger_42x-100.committed_prefix", + "../ledger_42-100.committed_prefix"}) + { + CHECK_FALSE(get_ledger_committed_prefix_range_from_file_name(invalid_name) + .has_value()); + } +} + +TEST_CASE("committed prefix files are ignored by the host ledger") +{ + const auto prefix = "ledger_42-100.committed_prefix"; + CHECK(is_ledger_file_name_committed_prefix(prefix)); + CHECK(is_ledger_file_ignored(prefix)); + CHECK_FALSE(is_ledger_file_name_committed(prefix)); +} + TEST_CASE("is_ledger_file_name_recovery: detects recovery suffix") { CHECK(is_ledger_file_name_recovery("ledger_1-100.committed.recovery")); diff --git a/src/host/test/ledger.cpp b/src/host/test/ledger.cpp index 608d37b6677f..5d08033462b3 100644 --- a/src/host/test/ledger.cpp +++ b/src/host/test/ledger.cpp @@ -161,6 +161,38 @@ void verify_framed_entries_range( REQUIRE(idx == read_result.end_idx + 1); } +void verify_completed_chunk( + const std::vector& chunk, size_t from, size_t to) +{ + const auto entry_count = to - from + 1; + const auto framed_entry_size = + ccf::kv::serialised_entry_header_size + sizeof(TestLedgerEntry); + const auto expected_positions_offset = + sizeof(size_t) + entry_count * framed_entry_size; + + const uint8_t* data = chunk.data(); + auto size = chunk.size(); + const auto positions_offset = serialized::read(data, size); + REQUIRE(positions_offset == expected_positions_offset); + REQUIRE(chunk.size() == positions_offset + entry_count * sizeof(uint32_t)); + + LedgerReadResult read_result{ + .data = std::vector( + chunk.begin() + sizeof(size_t), chunk.begin() + positions_offset), + .end_idx = to}; + verify_framed_entries_range(read_result, from, to); + + data = chunk.data() + positions_offset; + size = chunk.size() - positions_offset; + for (size_t i = 0; i < entry_count; ++i) + { + REQUIRE( + serialized::read(data, size) == + sizeof(size_t) + i * framed_entry_size); + } + REQUIRE(size == 0); +} + void read_entry_from_ledger(Ledger& ledger, size_t idx) { auto framed_entry = ledger.read_entry(idx); @@ -989,6 +1021,112 @@ TEST_CASE("Commit") } } +TEST_CASE("Committed ledger prefixes") +{ + auto dir = AutoDeleteFolder(ledger_dir); + + Ledger ledger(ledger_dir, wf); + TestEntrySubmitter entry_submitter(ledger, 1024); + + for (size_t i = 0; i < 9; ++i) + { + entry_submitter.write(true); + } + entry_submitter.write(true, ccf::kv::FORCE_LEDGER_CHUNK_AFTER); + + ledger.commit(5); + REQUIRE(number_of_committed_files_in_ledger_dir() == 0); + + const auto first_range = ledger.committed_ledger_prefix_range_with_idx(1); + REQUIRE(first_range.has_value()); + REQUIRE(first_range->first == 1); + REQUIRE(first_range->second == 5); + + const auto middle_range = ledger.committed_ledger_prefix_range_with_idx(3); + REQUIRE(middle_range.has_value()); + REQUIRE(middle_range->first == 3); + REQUIRE(middle_range->second == 5); + + REQUIRE_FALSE(ledger.committed_ledger_prefix_range_with_idx(0).has_value()); + REQUIRE_FALSE(ledger.committed_ledger_prefix_range_with_idx(6).has_value()); + REQUIRE_FALSE(ledger.read_committed_ledger_prefix(0, 5).has_value()); + REQUIRE_FALSE(ledger.read_committed_ledger_prefix(1, 6).has_value()); + REQUIRE_FALSE(ledger.read_committed_ledger_prefix(5, 4).has_value()); + + const auto first_prefix = ledger.read_committed_ledger_prefix(1, 5); + REQUIRE(first_prefix.has_value()); + verify_completed_chunk(first_prefix.value(), 1, 5); + + ledger.commit(8); + const auto second_range = ledger.committed_ledger_prefix_range_with_idx(6); + REQUIRE(second_range.has_value()); + REQUIRE(second_range->first == 6); + REQUIRE(second_range->second == 8); + + const auto second_prefix = ledger.read_committed_ledger_prefix(6, 8); + REQUIRE(second_prefix.has_value()); + verify_completed_chunk(second_prefix.value(), 6, 8); + + const auto first_prefix_again = ledger.read_committed_ledger_prefix(1, 5); + REQUIRE(first_prefix_again.has_value()); + REQUIRE(first_prefix_again.value() == first_prefix.value()); + + ledger.commit(10); + REQUIRE(number_of_committed_files_in_ledger_dir() == 1); + REQUIRE_FALSE(ledger.committed_ledger_prefix_range_with_idx(9).has_value()); + + const auto promoted_prefix = ledger.read_committed_ledger_prefix(1, 5); + REQUIRE(promoted_prefix.has_value()); + REQUIRE(promoted_prefix.value() == first_prefix.value()); +} + +TEST_CASE("Committed ledger prefix files are not recovered") +{ + auto dir = AutoDeleteFolder(ledger_dir); + auto ro_dir = AutoDeleteFolder(ledger_dir_read_only); + fs::create_directory(ledger_dir); + fs::create_directory(ledger_dir_read_only); + + std::vector prefix; + { + LedgerFile source(ledger_dir, 1); + for (size_t idx = 1; idx <= 5; ++idx) + { + const auto entry = make_ledger_entry(idx); + source.write_entry(entry.data(), entry.size(), true); + } + + const auto result = source.read_entries_as_completed_chunk(1, 5); + REQUIRE(result.has_value()); + prefix = std::move(result.value()); + } + + fs::remove_all(ledger_dir); + fs::create_directory(ledger_dir); + const auto prefix_path = fs::path(ledger_dir) / "ledger_1-5.committed_prefix"; + files::dump(prefix, prefix_path); + + { + Ledger ledger(ledger_dir, wf); + REQUIRE(ledger.get_last_idx() == 0); + } + REQUIRE_FALSE(fs::exists(prefix_path)); + REQUIRE( + fs::exists(fmt::format("{}{}", prefix_path, ledger_ignored_file_suffix))); + + const auto read_only_prefix_path = + fs::path(ledger_dir_read_only) / prefix_path.filename(); + files::dump(prefix, read_only_prefix_path); + Ledger ledger( + ledger_dir, + wf, + ledger_max_read_cache_files_default, + {ledger_dir_read_only}); + REQUIRE(ledger.get_last_idx() == 0); + REQUIRE_FALSE(ledger.read_entry(1).has_value()); + REQUIRE(fs::exists(read_only_prefix_path)); +} + TEST_CASE("Restore existing ledger") { auto dir = AutoDeleteFolder(ledger_dir); diff --git a/src/host/test/ledger_bench.cpp b/src/host/test/ledger_bench.cpp index f960ee497031..f1b439e7f437 100644 --- a/src/host/test/ledger_bench.cpp +++ b/src/host/test/ledger_bench.cpp @@ -16,6 +16,12 @@ namespace static constexpr size_t mebibyte = 1024 * kibibyte; static constexpr size_t entry_count = 10'000; + template + inline void do_not_optimize(const T& value) + { + asm volatile("" : : "g"(value) : "memory"); + } + struct RemoveDirectory { fs::path path; @@ -144,6 +150,53 @@ namespace state, "100_mib", prepare_sized_file<100 * mebibyte>); } + template + void benchmark_chunk_read( + picobench::state& state, const std::string& fixture_name) + { + const auto directory = fs::path( + fmt::format("ledger_read_bench_{}_{}", fixture_name, MaterializePrefix)); + fs::remove_all(directory); + fs::create_directory(directory); + RemoveDirectory remove_directory{directory}; + + LedgerFile file(directory, 1); + prepare_sized_file(file); + + state.start_timer(); + for ([[maybe_unused]] auto iteration : state) + { + if constexpr (MaterializePrefix) + { + const auto result = file.read_entries_as_completed_chunk(1, 1); + do_not_optimize(result); + } + else + { + const auto result = files::slurp((directory / "ledger_1").string()); + do_not_optimize(result); + } + } + state.stop_timer(); + } + +#define DECLARE_CHUNK_READ_BENCHMARKS(NAME, SIZE) \ + static void read_##NAME(picobench::state& state) \ + { \ + benchmark_chunk_read(state, #NAME); \ + } \ + static void materialize_##NAME##_prefix(picobench::state& state) \ + { \ + benchmark_chunk_read(state, #NAME); \ + } + + DECLARE_CHUNK_READ_BENCHMARKS(64_kib, 64 * kibibyte); + DECLARE_CHUNK_READ_BENCHMARKS(1_mib, mebibyte); + DECLARE_CHUNK_READ_BENCHMARKS(5_mib, 5 * mebibyte); + DECLARE_CHUNK_READ_BENCHMARKS(50_mib, 50 * mebibyte); + +#undef DECLARE_CHUNK_READ_BENCHMARKS + const std::vector rename_iterations = {20}; } @@ -167,6 +220,22 @@ PICOBENCH_SUITE("rename 100 MiB ledger file"); PICOBENCH(rename_100_mib).iterations(rename_iterations).baseline(); PICOBENCH(rename_100_mib_close_and_reopen).iterations(rename_iterations); +PICOBENCH_SUITE("read 64 KiB ledger chunk"); +PICOBENCH(read_64_kib).iterations({100}).baseline(); +PICOBENCH(materialize_64_kib_prefix).iterations({100}); + +PICOBENCH_SUITE("read 1 MiB ledger chunk"); +PICOBENCH(read_1_mib).iterations({20}).baseline(); +PICOBENCH(materialize_1_mib_prefix).iterations({20}); + +PICOBENCH_SUITE("read 5 MiB ledger chunk"); +PICOBENCH(read_5_mib).iterations({5}).baseline(); +PICOBENCH(materialize_5_mib_prefix).iterations({5}); + +PICOBENCH_SUITE("read 50 MiB ledger chunk"); +PICOBENCH(read_50_mib).iterations({1}).baseline(); +PICOBENCH(materialize_50_mib_prefix).iterations({1}); + int main(int argc, char* argv[]) { ccf::logger::config::level() = ccf::LoggerLevel::FATAL; diff --git a/src/node/rpc/file_serving_handlers.h b/src/node/rpc/file_serving_handlers.h index cc74de27eca1..845713791209 100644 --- a/src/node/rpc/file_serving_handlers.h +++ b/src/node/rpc/file_serving_handlers.h @@ -150,7 +150,7 @@ namespace ccf::node return node_configuration_subsystem; } - // Helper function to serve byte ranges from a file stream. + // Helper function to serve byte ranges from a resource. // This populates the response body, and range-related response headers. This // may produce an error response if an invalid range was requested. // @@ -169,13 +169,13 @@ namespace ccf::node // with any other metadata headers) _before_ calling this function, and // generally avoid modifying the response further _after_ calling this // function. + template // NOLINTNEXTLINE(readability-function-cognitive-complexity) - static void fill_range_response_from_file( - ccf::endpoints::CommandEndpointContext& ctx, std::ifstream& f) + static void fill_range_response( + ccf::endpoints::CommandEndpointContext& ctx, + size_t total_size, + const ReadRange& read_range) { - f.seekg(0, std::ifstream::end); - const auto total_size = (size_t)f.tellg(); - if (total_size == 0) { // Refuse to return an empty file - it's not going to be a valid snapshot @@ -349,8 +349,17 @@ namespace ccf::node return; } + if (offset == 0) + { + ctx.rpc_ctx->set_error( + HTTP_STATUS_BAD_REQUEST, + ccf::errors::InvalidHeaderValue, + "Range suffix length must be greater than zero"); + return; + } + range_end = total_size; - range_start = range_end - offset; + range_start = offset >= range_end ? 0 : range_end - offset; } else { @@ -379,15 +388,8 @@ namespace ccf::node std::vector contents; if (digest_algo.has_value()) { - // Need full file contents for the digest - f.seekg(0); - std::vector full_contents(total_size); - // NOLINTNEXTLINE(cppcoreguidelines-pro-type-reinterpret-cast) - f.read(reinterpret_cast(full_contents.data()), total_size); - f.close(); - - auto bytes_read = static_cast(f.gcount()); - if (bytes_read < range_end) + auto full_contents = read_range(0, total_size); + if (!full_contents.has_value() || full_contents->size() != total_size) { ctx.rpc_ctx->set_error( HTTP_STATUS_INTERNAL_SERVER_ERROR, @@ -396,29 +398,38 @@ namespace ccf::node return; } - if (bytes_read == total_size) - { - ctx.rpc_ctx->set_response_header( - ccf::http::headers::REPR_DIGEST, - format_repr_digest( - digest_algo->first, - digest_algo->second, - full_contents.data(), - full_contents.size())); - } + ctx.rpc_ctx->set_response_header( + ccf::http::headers::REPR_DIGEST, + format_repr_digest( + digest_algo->first, + digest_algo->second, + full_contents->data(), + full_contents->size())); // Extract the requested range - contents.assign( - full_contents.begin() + range_start, full_contents.begin() + range_end); + if (range_start == 0 && range_end == total_size) + { + contents = std::move(full_contents.value()); + } + else + { + contents.assign( + full_contents->begin() + range_start, + full_contents->begin() + range_end); + } } else { - // Read only the requested range - contents.resize(range_size); - f.seekg(range_start); - // NOLINTNEXTLINE(cppcoreguidelines-pro-type-reinterpret-cast) - f.read(reinterpret_cast(contents.data()), contents.size()); - f.close(); + auto range_contents = read_range(range_start, range_end); + if (!range_contents.has_value() || range_contents->size() != range_size) + { + ctx.rpc_ctx->set_error( + HTTP_STATUS_INTERNAL_SERVER_ERROR, + ccf::errors::InternalError, + "Server was unable to read the file correctly"); + return; + } + contents = std::move(range_contents.value()); } // Compute ETag over the response content (RFC 9530 structured field @@ -515,11 +526,74 @@ namespace ccf::node } } + static void fill_range_response_from_file( + ccf::endpoints::CommandEndpointContext& ctx, std::ifstream& f) + { + f.seekg(0, std::ifstream::end); + const auto end = f.tellg(); + if (end < 0) + { + ctx.rpc_ctx->set_error( + HTTP_STATUS_INTERNAL_SERVER_ERROR, + ccf::errors::InternalError, + "Server was unable to determine the file size"); + return; + } + + const auto total_size = static_cast(end); + const auto read_range = + [&f](size_t start, size_t end) -> std::optional> { + const auto size = end - start; + std::vector contents(size); + f.clear(); + f.seekg(static_cast(start), std::ifstream::beg); + if (!f.good()) + { + return std::nullopt; + } + + // NOLINTNEXTLINE(cppcoreguidelines-pro-type-reinterpret-cast) + f.read(reinterpret_cast(contents.data()), contents.size()); + if (static_cast(f.gcount()) != size) + { + return std::nullopt; + } + return contents; + }; + + fill_range_response(ctx, total_size, read_range); + } + + static void fill_range_response_from_contents( + ccf::endpoints::CommandEndpointContext& ctx, std::vector&& source) + { + const auto total_size = source.size(); + const auto read_range = + [&source]( + size_t start, size_t end) -> std::optional> { + if (start > end || end > source.size()) + { + return std::nullopt; + } + + if (start == 0 && end == source.size()) + { + return std::move(source); + } + + return std::vector(source.begin() + start, source.begin() + end); + }; + + fill_range_response(ctx, total_size, read_range); + } + // NOLINTNEXTLINE(readability-function-cognitive-complexity) static void init_file_serving_handlers( ccf::BaseEndpointRegistry& registry, ccf::AbstractNodeContext& node_context) { static constexpr auto file_since_param_key = "since"; + static constexpr auto include_committed_prefix_param_key = + "include_committed_prefix"; auto find_snapshot = [&](ccf::endpoints::ReadOnlyEndpointContext& ctx) { size_t latest_idx = 0; @@ -659,8 +733,8 @@ namespace ccf::node // Find a ledger chunk that includes the since value auto find_chunk = [&](ccf::endpoints::ReadOnlyEndpointContext& ctx) { size_t since_idx = 0; + bool include_committed_prefix = false; { - // Get since_idx from query param, if present const auto parsed_query = http::parse_query(ctx.rpc_ctx->get_request_query()); @@ -689,6 +763,23 @@ namespace ccf::node "Missing required query parameter '{}'", file_since_param_key)); return; } + + if (parsed_query.contains(include_committed_prefix_param_key)) + { + error_reason.clear(); + if (!http::get_query_value( + parsed_query, + include_committed_prefix_param_key, + include_committed_prefix, + error_reason)) + { + ctx.rpc_ctx->set_error( + HTTP_STATUS_BAD_REQUEST, + ccf::errors::InvalidQueryParameterValue, + std::move(error_reason)); + return; + } + } } LOG_DEBUG_FMT("Finding ledger chunk including index {}", since_idx); @@ -723,20 +814,63 @@ namespace ccf::node const auto chunk_path = read_ledger_subsystem->committed_ledger_path_with_idx(since_idx); + const auto redirect_to_committed_chunk = + [&](const std::filesystem::path& path) { + const auto chunk_filename = path.filename(); + const auto redirect_url = fmt::format( + "https://{}/node/ledger_chunk/{}", address.value(), chunk_filename); + LOG_DEBUG_FMT("Redirecting to ledger chunk: {}", redirect_url); + ctx.rpc_ctx->set_response_header( + ccf::http::headers::LOCATION, redirect_url); + ctx.rpc_ctx->set_response_status(HTTP_STATUS_PERMANENT_REDIRECT); + }; + // If the file is found locally, always serve it from this node if (chunk_path.has_value()) { - const auto chunk_filename = chunk_path.value().filename(); - - auto redirect_url = fmt::format( - "https://{}/node/ledger_chunk/{}", address.value(), chunk_filename); - LOG_DEBUG_FMT("Redirecting to ledger chunk: {}", redirect_url); - ctx.rpc_ctx->set_response_header( - ccf::http::headers::LOCATION, redirect_url); - ctx.rpc_ctx->set_response_status(HTTP_STATUS_PERMANENT_REDIRECT); + redirect_to_committed_chunk(chunk_path.value()); return; } + if (include_committed_prefix) + { + const auto prefix_range = + read_ledger_subsystem->committed_ledger_prefix_range_with_idx( + since_idx); + if (prefix_range.has_value()) + { + const auto chunk_filename = fmt::format( + "ledger_{}-{}{}", + prefix_range->start_idx, + prefix_range->end_idx, + asynchost::ledger_committed_prefix_suffix); + const auto redirect_url = fmt::format( + "https://{}/node/ledger_chunk/committed_prefix/{}", + address.value(), + chunk_filename); + LOG_DEBUG_FMT( + "Redirecting to committed ledger prefix: {}", redirect_url); + ctx.rpc_ctx->set_response_header( + ccf::http::headers::LOCATION, redirect_url); + ctx.rpc_ctx->set_response_header( + ccf::http::headers::CACHE_CONTROL, "no-store"); + ctx.rpc_ctx->set_response_status(HTTP_STATUS_TEMPORARY_REDIRECT); + return; + } + + // A completed source file may have been promoted after the first + // committed-file lookup but before the prefix lookup. Retry the + // canonical lookup so this atomic state transition cannot produce a + // false 404. + const auto promoted_chunk_path = + read_ledger_subsystem->committed_ledger_path_with_idx(since_idx); + if (promoted_chunk_path.has_value()) + { + redirect_to_committed_chunk(promoted_chunk_path.value()); + return; + } + } + // Otherwise, if the file is before our init index, i.e. where we started // replicating, redirect to the next node in order. const size_t init_idx = read_ledger_subsystem->get_init_idx(); @@ -760,9 +894,17 @@ namespace ccf::node address.value(), file_since_param_key, since_idx); + if (include_committed_prefix) + { + location += + fmt::format("&{}=true", include_committed_prefix_param_key); + ctx.rpc_ctx->set_response_header( + ccf::http::headers::CACHE_CONTROL, "no-store"); + } ctx.rpc_ctx->set_response_header(http::headers::LOCATION, location); ctx.rpc_ctx->set_error( - HTTP_STATUS_PERMANENT_REDIRECT, + include_committed_prefix ? HTTP_STATUS_TEMPORARY_REDIRECT : + HTTP_STATUS_PERMANENT_REDIRECT, ccf::errors::NodeCannotHandleRequest, "Node does not have ledger chunk; redirecting to next node"); return; @@ -785,10 +927,18 @@ namespace ccf::node auto location = fmt::format("https://{}/node/ledger_chunk", address.value()); location += fmt::format("?{}={}", file_since_param_key, since_idx); + if (include_committed_prefix) + { + location += + fmt::format("&{}=true", include_committed_prefix_param_key); + ctx.rpc_ctx->set_response_header( + ccf::http::headers::CACHE_CONTROL, "no-store"); + } ctx.rpc_ctx->set_response_header(http::headers::LOCATION, location); ctx.rpc_ctx->set_error( - HTTP_STATUS_PERMANENT_REDIRECT, + include_committed_prefix ? HTTP_STATUS_TEMPORARY_REDIRECT : + HTTP_STATUS_PERMANENT_REDIRECT, ccf::errors::NodeCannotHandleRequest, fmt::format( "Ledger chunk including index {} not found locally; " @@ -800,6 +950,11 @@ namespace ccf::node } // Redirect possibilities exhausted + if (include_committed_prefix) + { + ctx.rpc_ctx->set_response_header( + ccf::http::headers::CACHE_CONTROL, "no-store"); + } ctx.rpc_ctx->set_error( HTTP_STATUS_NOT_FOUND, ccf::errors::ResourceNotFound, @@ -813,6 +968,11 @@ namespace ccf::node .set_forwarding_required(endpoints::ForwardingRequired::Never) .add_query_parameter( file_since_param_key, ccf::endpoints::RequiredParameter) + .add_query_parameter( + include_committed_prefix_param_key, ccf::endpoints::OptionalParameter) + .add_openapi_response( + HTTP_STATUS_TEMPORARY_REDIRECT, + "Redirect to a temporary committed ledger prefix.") .add_openapi_response( HTTP_STATUS_PERMANENT_REDIRECT, "Redirect to the selected ledger chunk.") @@ -823,7 +983,9 @@ namespace ccf::node .set_openapi_description( "Redirect to the corresponding /node/ledger_chunk/{chunk_name} " "endpoint for the ledger chunk including the sequence number specified " - "in the 'since' query parameter.") + "in the 'since' query parameter. If 'include_committed_prefix' is true " + "and no committed file is available, this may temporarily redirect to " + "a synthetic committed-prefix resource.") .install(); registry .make_read_only_endpoint( @@ -831,6 +993,11 @@ namespace ccf::node .set_forwarding_required(endpoints::ForwardingRequired::Never) .add_query_parameter( file_since_param_key, ccf::endpoints::RequiredParameter) + .add_query_parameter( + include_committed_prefix_param_key, ccf::endpoints::OptionalParameter) + .add_openapi_response( + HTTP_STATUS_TEMPORARY_REDIRECT, + "Redirect to a temporary committed ledger prefix.") .add_openapi_response( HTTP_STATUS_PERMANENT_REDIRECT, "Redirect to the selected ledger chunk.") @@ -841,7 +1008,9 @@ namespace ccf::node .set_openapi_description( "Redirect to the corresponding /node/ledger_chunk/{chunk_name} " "endpoint for the ledger chunk including the sequence number specified " - "in the 'since' query parameter.") + "in the 'since' query parameter. If 'include_committed_prefix' is true " + "and no committed file is available, this may temporarily redirect to " + "a synthetic committed-prefix resource.") .install(); auto get_snapshot = [&](ccf::endpoints::CommandEndpointContext& ctx) { @@ -1018,5 +1187,118 @@ namespace ccf::node "Download a specific ledger chunk by name. Supports HTTP Range header " "for partial downloads.") .install(); + + auto get_committed_ledger_prefix = + [&](ccf::endpoints::CommandEndpointContext& ctx) { + ctx.rpc_ctx->set_response_header( + ccf::http::headers::CACHE_CONTROL, "no-store"); + + std::string chunk_name; + std::string error; + if (!ccf::endpoints::get_path_param( + ctx.rpc_ctx->get_request_path_params(), + "chunk_name", + chunk_name, + error)) + { + ctx.rpc_ctx->set_error( + HTTP_STATUS_BAD_REQUEST, + ccf::errors::InvalidResourceName, + std::move(error)); + return; + } + + const auto range = + asynchost::get_ledger_committed_prefix_range_from_file_name( + chunk_name); + if (!range.has_value()) + { + ctx.rpc_ctx->set_error( + HTTP_STATUS_BAD_REQUEST, + ccf::errors::InvalidResourceName, + fmt::format( + "{} is not a valid committed ledger prefix name", chunk_name)); + return; + } + + auto read_ledger_subsystem = + node_context.get_subsystem(); + if (read_ledger_subsystem == nullptr) + { + ctx.rpc_ctx->set_error( + HTTP_STATUS_INTERNAL_SERVER_ERROR, + ccf::errors::InternalError, + "LedgerReadSubsystem is not available"); + return; + } + + auto contents = read_ledger_subsystem->read_committed_ledger_prefix( + range->first, range->second); + if (!contents.has_value()) + { + ctx.rpc_ctx->set_error( + HTTP_STATUS_NOT_FOUND, + ccf::errors::ResourceNotFound, + fmt::format( + "This node cannot provide committed ledger prefix {}", + chunk_name)); + return; + } + + ctx.rpc_ctx->set_response_header( + ccf::http::headers::CCF_LEDGER_CHUNK_NAME, chunk_name); + ctx.rpc_ctx->set_response_header( + ccf::http::headers::CCF_LEDGER_CHUNK_KIND, "committed-prefix"); + fill_range_response_from_contents(ctx, std::move(contents.value())); + }; + registry + .make_command_endpoint( + "/ledger_chunk/committed_prefix/{chunk_name}", + HTTP_HEAD, + get_committed_ledger_prefix, + no_auth_required) + .set_forwarding_required(endpoints::ForwardingRequired::Never) + .add_openapi_response( + HTTP_STATUS_OK, "Metadata for the requested committed ledger prefix.") + .add_openapi_response( + HTTP_STATUS_PARTIAL_CONTENT, + "Metadata for the requested committed ledger prefix range.") + .add_openapi_response( + HTTP_STATUS_NOT_MODIFIED, + "The requested committed ledger prefix has not changed.") + .add_openapi_response( + HTTP_STATUS_NOT_FOUND, + "The requested committed ledger prefix is not available.") + .require_operator_feature(endpoints::OperatorFeature::LedgerChunkRead) + .set_openapi_summary("Committed ledger prefix metadata") + .set_openapi_description( + "Metadata about a synthetic chunk containing only committed ledger " + "entries. The resource is not a canonical .committed ledger file.") + .install(); + registry + .make_command_endpoint( + "/ledger_chunk/committed_prefix/{chunk_name}", + HTTP_GET, + get_committed_ledger_prefix, + no_auth_required) + .set_forwarding_required(endpoints::ForwardingRequired::Never) + .add_openapi_response( + HTTP_STATUS_OK, "The requested committed ledger prefix.") + .add_openapi_response( + HTTP_STATUS_PARTIAL_CONTENT, + "The requested byte range of the committed ledger prefix.") + .add_openapi_response( + HTTP_STATUS_NOT_MODIFIED, + "The requested committed ledger prefix has not changed.") + .add_openapi_response( + HTTP_STATUS_NOT_FOUND, + "The requested committed ledger prefix is not available.") + .require_operator_feature(endpoints::OperatorFeature::LedgerChunkRead) + .set_openapi_summary("Download committed ledger prefix") + .set_openapi_description( + "Download a synthetic chunk containing only committed ledger entries. " + "Supports HTTP Range and digest headers. The resource is not a " + "canonical .committed ledger file and must not be used for recovery.") + .install(); } } \ No newline at end of file diff --git a/src/node/rpc/ledger_interface.h b/src/node/rpc/ledger_interface.h index e1531fd205b1..f66091f14711 100644 --- a/src/node/rpc/ledger_interface.h +++ b/src/node/rpc/ledger_interface.h @@ -6,9 +6,16 @@ #include #include +#include namespace ccf { + struct CommittedLedgerPrefixRange + { + size_t start_idx; + size_t end_idx; + }; + class AbstractReadLedgerSubsystemInterface : public AbstractNodeSubSystem { public: @@ -22,6 +29,12 @@ namespace ccf virtual std::optional committed_ledger_path_with_idx( size_t idx) = 0; + virtual std::optional + committed_ledger_prefix_range_with_idx(size_t idx) = 0; + + virtual std::optional> read_committed_ledger_prefix( + size_t from, size_t to) = 0; + virtual size_t get_init_idx() = 0; }; } \ No newline at end of file diff --git a/src/node/rpc/ledger_subsystem.h b/src/node/rpc/ledger_subsystem.h index d2ee151999a7..f83a0d68bc71 100644 --- a/src/node/rpc/ledger_subsystem.h +++ b/src/node/rpc/ledger_subsystem.h @@ -21,6 +21,25 @@ namespace ccf return ledger.committed_ledger_path_with_idx(idx); } + [[nodiscard]] std::optional + committed_ledger_prefix_range_with_idx(size_t idx) override + { + const auto range = ledger.committed_ledger_prefix_range_with_idx(idx); + if (!range.has_value()) + { + return std::nullopt; + } + + return CommittedLedgerPrefixRange{ + .start_idx = range->first, .end_idx = range->second}; + } + + [[nodiscard]] std::optional> + read_committed_ledger_prefix(size_t from, size_t to) override + { + return ledger.read_committed_ledger_prefix(from, to); + } + [[nodiscard]] size_t get_init_idx() override { return ledger.get_init_idx(); diff --git a/src/node/rpc/node_frontend.h b/src/node/rpc/node_frontend.h index 2f90dc0e26a3..77de1065e7d7 100644 --- a/src/node/rpc/node_frontend.h +++ b/src/node/rpc/node_frontend.h @@ -444,7 +444,7 @@ namespace ccf openapi_info.description = "This API provides public, uncredentialed access to service and node " "state."; - openapi_info.document_version = "5.0.6"; + openapi_info.document_version = "5.0.7"; } // NOLINTNEXTLINE(readability-function-cognitive-complexity) diff --git a/tests/e2e_operations.py b/tests/e2e_operations.py index b93834c91635..742a7a4ba6d9 100644 --- a/tests/e2e_operations.py +++ b/tests/e2e_operations.py @@ -1296,6 +1296,22 @@ def test_ledger_chunk_access(network, args): range_etag == f'"sha-256=:{range_expected_b64}:"' ), f"Range ETag mismatch: expected sha-256=:{range_expected_b64}:, got {range_etag}" + # A suffix longer than the representation selects the full content. + r = c.get( + chunk_url, + headers={"range": f"bytes=-{total_size + 1}"}, + allow_redirects=False, + ) + assert r.status_code == http.HTTPStatus.PARTIAL_CONTENT.value, r + assert r.body.data() == chunk_data + + r = c.get( + chunk_url, + headers={"range": "bytes=-0"}, + allow_redirects=False, + ) + assert r.status_code == http.HTTPStatus.BAD_REQUEST.value, r + # Non-matching If-None-Match on range -> fresh partial download r = c.call( chunk_url, @@ -1484,6 +1500,140 @@ def test_ledger_chunk_repr_digest(network, args): ), f"Expected partial body of {range_end + 1} bytes, got {len(r.body.data())}" +def test_committed_ledger_prefix_access(network, args): + primary, _ = network.find_primary() + first_txid = network.txs.issue( + network, number_txs=1, send_private=False, send_public=True + ) + + with primary.client( + interface_name=infra.interfaces.FILE_SERVING_RPC_INTERFACE + ) as c: + locator = f"/node/ledger_chunk?since={first_txid.seqno}" + + r = c.get(locator, allow_redirects=False) + assert r.status_code == http.HTTPStatus.NOT_FOUND, r + + r = c.get(f"{locator}&include_committed_prefix=false", allow_redirects=False) + assert r.status_code == http.HTTPStatus.NOT_FOUND, r + + r = c.get(f"{locator}&include_committed_prefix=invalid", allow_redirects=False) + assert r.status_code == http.HTTPStatus.BAD_REQUEST, r + + r = c.head(f"{locator}&include_committed_prefix=true", allow_redirects=False) + assert r.status_code == http.HTTPStatus.TEMPORARY_REDIRECT, r + assert r.headers["cache-control"] == "no-store", r + + r = c.get(f"{locator}&include_committed_prefix=true", allow_redirects=False) + assert r.status_code == http.HTTPStatus.TEMPORARY_REDIRECT, r + assert r.headers["cache-control"] == "no-store", r + prefix_url = urllib.parse.urlparse(r.headers["Location"]).path + first_prefix_name = os.path.basename(prefix_url) + assert first_prefix_name.endswith( + ccf.ledger.COMMITTED_PREFIX_FILE_SUFFIX + ), first_prefix_name + first_start, first_end = ccf.ledger.range_from_filename(first_prefix_name) + assert first_start == first_txid.seqno + assert first_end is not None and first_end >= first_start + + r = c.head(prefix_url, allow_redirects=False) + assert r.status_code == http.HTTPStatus.OK, r + assert r.headers["cache-control"] == "no-store", r + assert r.headers["x-ms-ccf-ledger-chunk-kind"] == "committed-prefix", r + assert r.headers["x-ms-ccf-ledger-chunk-name"] == first_prefix_name, r + prefix_size = int(r.headers["content-length"]) + first_etag = r.headers["etag"] + + r = c.get( + prefix_url, + allow_redirects=False, + headers={"want-repr-digest": "sha-256=1"}, + ) + assert r.status_code == http.HTTPStatus.OK, r + assert r.headers["cache-control"] == "no-store", r + first_prefix = r.body.data() + assert len(first_prefix) == prefix_size + expected_digest = base64.b64encode( + hashlib.sha256(first_prefix).digest() + ).decode() + assert r.headers["repr-digest"] == f"sha-256=:{expected_digest}:" + + range_end = min(31, prefix_size - 1) + r = c.get( + prefix_url, + allow_redirects=False, + headers={"range": f"bytes=0-{range_end}"}, + ) + assert r.status_code == http.HTTPStatus.PARTIAL_CONTENT, r + assert r.headers["cache-control"] == "no-store", r + assert r.body.data() == first_prefix[: range_end + 1] + + with tempfile.TemporaryDirectory() as ledger_dir: + first_path = os.path.join(ledger_dir, first_prefix_name) + with open(first_path, "wb") as prefix_file: + prefix_file.write(first_prefix) + + chunk = ccf.ledger.LedgerChunk(first_path) + assert chunk.get_seqnos() == (first_start, first_end) + assert len(chunk) == first_end - first_start + 1 + assert not chunk.is_committed() + assert len(ccf.ledger.Ledger([ledger_dir])) == 0 + + second_txid = network.txs.issue( + network, number_txs=1, send_private=False, send_public=True + ) + assert second_txid.seqno > first_end + + r = c.get(prefix_url, allow_redirects=False) + assert r.status_code == http.HTTPStatus.OK, r + assert r.body.data() == first_prefix + assert r.headers["etag"] == first_etag + + next_seqno = first_end + 1 + r = c.get( + "/node/ledger_chunk" + f"?since={next_seqno}&include_committed_prefix=true", + allow_redirects=False, + ) + assert r.status_code == http.HTTPStatus.TEMPORARY_REDIRECT, r + assert r.headers["cache-control"] == "no-store", r + second_url = urllib.parse.urlparse(r.headers["Location"]).path + second_prefix_name = os.path.basename(second_url) + second_start, second_end = ccf.ledger.range_from_filename( + second_prefix_name + ) + assert second_start == next_seqno + assert second_end is not None and second_end >= second_txid.seqno + + r = c.get(second_url, allow_redirects=False) + assert r.status_code == http.HTTPStatus.OK, r + second_path = os.path.join(ledger_dir, second_prefix_name) + with open(second_path, "wb") as prefix_file: + prefix_file.write(r.body.data()) + + ledger = ccf.ledger.Ledger([ledger_dir], committed_only=False) + assert len(ledger) == 2 + assert [chunk.get_seqnos() for chunk in ledger] == [ + (first_start, first_end), + (second_start, second_end), + ] + + r = c.get( + "/node/ledger_chunk" + f"?since={second_end + 1_000_000}&include_committed_prefix=true", + allow_redirects=False, + ) + assert r.status_code == http.HTTPStatus.NOT_FOUND, r + assert r.headers["cache-control"] == "no-store", r + + r = c.get( + "/node/ledger_chunk/committed_prefix/not-a-prefix", + allow_redirects=False, + ) + assert r.status_code == http.HTTPStatus.BAD_REQUEST, r + assert r.headers["cache-control"] == "no-store", r + + def test_ledger_chunk_redirect_recent(network, args): """ Access ledger chunk that is missing locally on a backup, after the initial index, @@ -1506,10 +1656,6 @@ def test_ledger_chunk_redirect_recent(network, args): with late_backup.client( interface_name=infra.interfaces.FILE_SERVING_RPC_INTERFACE ) as c: - r = c.head( - f"/node/ledger_chunk?since={start_of_last_chunk}", allow_redirects=False - ) - assert r.status_code == http.HTTPStatus.PERMANENT_REDIRECT.value, r expected_host = primary.get_public_rpc_host( interface_name=infra.interfaces.FILE_SERVING_RPC_INTERFACE ) @@ -1517,6 +1663,23 @@ def test_ledger_chunk_redirect_recent(network, args): interface_name=infra.interfaces.FILE_SERVING_RPC_INTERFACE ) expected_location = f"https://{infra.interfaces.make_address(expected_host, expected_port)}/node/ledger_chunk?since={start_of_last_chunk}" + + r = c.head( + f"/node/ledger_chunk?since={start_of_last_chunk}" + "&include_committed_prefix=true", + allow_redirects=False, + ) + assert r.status_code == http.HTTPStatus.TEMPORARY_REDIRECT.value, r + assert r.headers["cache-control"] == "no-store", r + assert ( + r.headers["Location"] + == f"{expected_location}&include_committed_prefix=true" + ), r + + r = c.head( + f"/node/ledger_chunk?since={start_of_last_chunk}", allow_redirects=False + ) + assert r.status_code == http.HTTPStatus.PERMANENT_REDIRECT.value, r assert r.headers["Location"] == expected_location, r r = c.get( f"/node/ledger_chunk?since={start_of_last_chunk}", allow_redirects=True @@ -1582,6 +1745,21 @@ def test_ledger_chunk_redirect_gap(network, args): with new_node.client( interface_name=infra.interfaces.FILE_SERVING_RPC_INTERFACE ) as c: + r = c.head( + f"/node/ledger_chunk?since={download_index}" + "&include_committed_prefix=true", + allow_redirects=False, + ) + assert r.status_code == http.HTTPStatus.TEMPORARY_REDIRECT.value, r + assert r.headers["cache-control"] == "no-store", r + redirect_query = urllib.parse.parse_qs( + urllib.parse.urlparse(r.headers["Location"]).query + ) + assert redirect_query == { + "since": [str(download_index)], + "include_committed_prefix": ["true"], + }, r + r = c.head( f"/node/ledger_chunk?since={download_index}", allow_redirects=False ) @@ -1694,6 +1872,18 @@ def run_ledger_chunk_download(args): test_ledger_chunk_redirect_gap(network, args) +def run_committed_ledger_prefix_download(args): + with infra.network.network( + args.nodes, + args.binary_dir, + args.debug_nodes, + pdb=args.pdb, + txs=app.LoggingTxs("user0"), + ) as network: + network.start_and_open(args) + test_committed_ledger_prefix_access(network, args) + + def run_tls_san_checks(const_args): args = copy.deepcopy(const_args) args.label += "_tls_san" diff --git a/tests/schema.py b/tests/schema.py index 331a64c5cd53..0f90602524de 100644 --- a/tests/schema.py +++ b/tests/schema.py @@ -282,6 +282,15 @@ def add(parser): ledger_chunk_bytes="1B", # Chunk ledger at every signature transaction ) + cr.add( + "committed-prefix-download", + e2e_operations.run_committed_ledger_prefix_download, + package="samples/apps/logging/logging", + nodes=infra.e2e_args.min_nodes(cr.args, f=0), + initial_user_count=1, + ledger_chunk_bytes="50MB", + ) + cr.add( "download-snapshot", e2e_operations.run_backup_snapshot_download,