From 18fc9e658afce046b76e363dcb2bbbe1c7fd8649 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?=E8=B4=BE=E5=B2=9B?= Date: Fri, 6 Nov 2026 19:46:32 +0800 Subject: [PATCH 1/4] Add managed local blob cache import to ublkd MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Allow local producers to populate the existing Photon file cache after publishing an immutable blob, avoiding a redundant registry download. Validate the source size and SHA-256 before importing through the shared cache object so existing capacity, LRU and disk-space policies apply. Document the root-only endpoint and add a kernel-independent CTest for local reuse, request validation, automatic eviction and source refill. Signed-off-by: 贾岛 --- docs/standalone-usage.md | 17 +++ src/ublk/CMakeLists.txt | 1 + src/ublk/cache_import.cpp | 122 +++++++++++++++++ src/ublk/cache_import.h | 26 ++++ src/ublk/test/CMakeLists.txt | 18 +++ src/ublk/test/cache_import_test.cpp | 202 ++++++++++++++++++++++++++++ src/ublk/ublkd_main.cpp | 5 +- 7 files changed, 390 insertions(+), 1 deletion(-) create mode 100644 src/ublk/cache_import.cpp create mode 100644 src/ublk/cache_import.h create mode 100644 src/ublk/test/cache_import_test.cpp diff --git a/docs/standalone-usage.md b/docs/standalone-usage.md index 099b5d09..c2cd0c5b 100644 --- a/docs/standalone-usage.md +++ b/docs/standalone-usage.md @@ -111,6 +111,23 @@ exclusively across the daemon and CLI processes; devices owned by the daemon must be deleted through its API (the CLI `del` refuses them). A systemd unit template is installed at /opt/overlaybd/overlaybd-ublkd.service. +#### Import a local blob into the managed cache + +After publishing an immutable blob to a registry, a local producer can seed the daemon's existing file cache without downloading the same bytes again: + +```bash +BLOB=/absolute/path/to/finalized-blob +DIGEST=sha256:$(sha256sum "$BLOB" | cut -d ' ' -f 1) +SIZE=$(stat -c %s "$BLOB") +curl --unix-socket "$SOCK" -X POST -H 'Content-Type: application/json' \ + -d "{\"path\":\"$BLOB\",\"url\":\"https://registry.example/v2/repository/blobs/$DIGEST\",\"digest\":\"$DIGEST\",\"size\":$SIZE}" \ + http://d/v1/cache/import +``` + +The source must be a regular file visible in the daemon's mount namespace; the final path component cannot be a symlink. Keep its contents immutable until the request completes. The daemon verifies the positive size and SHA-256 before writing any cache bytes. The URL must be an HTTP(S) registry blob URL ending in `/blobs/` and should match the URL used by subsequent image reads. This root-only local operation does not publish the blob or authenticate to the registry; publish durable content before importing it. + +The endpoint returns `200` on completion, `400` for invalid requests or local content, `409` when the managed file cache is unavailable or another cache type is configured, and `503` if cache population cannot complete. Import is an optional optimization: failures, partial imports, and later eviction retain the normal source-read path. Producers may remove their staging files after the request finishes. Cache imports use the same capacity, LRU eviction, allocation accounting, and free-space protection as downloaded blobs; they do not pin entries or create a separate retained copy. Existing active device and mount lifecycles are unchanged. + ### Notes * Always unmount filesystems on `/dev/ublkbN` before `del` (or before stopping diff --git a/src/ublk/CMakeLists.txt b/src/ublk/CMakeLists.txt index f6c65f99..606182ad 100644 --- a/src/ublk/CMakeLists.txt +++ b/src/ublk/CMakeLists.txt @@ -7,6 +7,7 @@ add_library(ublk_frontend_lib cli.cpp config_patch.cpp ublkd_protocol.cpp + cache_import.cpp pool_placeholder.cpp blkdev_hygiene.cpp ) diff --git a/src/ublk/cache_import.cpp b/src/ublk/cache_import.cpp new file mode 100644 index 00000000..b514de1c --- /dev/null +++ b/src/ublk/cache_import.cpp @@ -0,0 +1,122 @@ +/* + Copyright The Overlaybd Authors + + Licensed under the Apache License, Version 2.0 (the "License"); + you may not use this file except in compliance with the License. + You may obtain a copy of the License at + + http://www.apache.org/licenses/LICENSE-2.0 + + Unless required by applicable law or agreed to in writing, software + distributed under the License is distributed on an "AS IS" BASIS, + WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + See the License for the specific language governing permissions and + limitations under the License. +*/ + +#include "cache_import.h" +#include "ublkd_protocol.h" +#include "../image_service.h" +#include +#include +#include + +#include +#include +#include +#include +#include +#include + +void ublkd_handle_cache_import(ImageService *service, const std::string &body, int &code, + std::string &msg) { + auto fail = [&](int status, const char *reason) { + code = status; + msg = ublkd_msg_error(reason); + }; + rapidjson::Document doc; + doc.Parse(body.c_str()); + if (doc.HasParseError() || !doc.IsObject() || !doc.HasMember("path") || + !doc["path"].IsString() || !doc.HasMember("url") || !doc["url"].IsString() || + !doc.HasMember("digest") || !doc["digest"].IsString() || !doc.HasMember("size") || + !doc["size"].IsInt64() || doc["size"].GetInt64() <= 0) { + fail(400, "expected path, url, sha256 digest and positive size"); + return; + } + std::string path(doc["path"].GetString(), doc["path"].GetStringLength()); + std::string url(doc["url"].GetString(), doc["url"].GetStringLength()); + std::string digest(doc["digest"].GetString(), doc["digest"].GetStringLength()); + const int64_t size = doc["size"].GetInt64(); + const std::string suffix = "/blobs/" + digest; + if (digest.size() != 71 || digest.substr(0, 7) != "sha256:" || + digest.find_first_not_of("0123456789abcdef", 7) != std::string::npos || path.empty() || + path[0] != '/' || path.find('\0') != std::string::npos || + url.find('\0') != std::string::npos || + (url.substr(0, 7) != "http://" && url.substr(0, 8) != "https://") || + url.size() <= suffix.size() || + url.compare(url.size() - suffix.size(), suffix.size(), suffix) != 0) { + fail(400, "invalid content-addressed cache import"); + return; + } + if (service->global_conf.cacheConfig().cacheType() != "file" || !service->global_fs.cached_fs) { + fail(409, "local import requires the managed file cache"); + return; + } + int fd = ::open(path.c_str(), O_RDONLY | O_CLOEXEC | O_NOFOLLOW | O_NONBLOCK); + if (fd < 0) { + fail(400, "cannot open local blob"); + return; + } + DEFER(::close(fd)); + struct stat st; + if (::fstat(fd, &st) != 0 || !S_ISREG(st.st_mode) || st.st_size != size) { + fail(400, "local blob size or type mismatch"); + return; + } + // Validate the whole file before exposing any bytes under its digest. + std::vector buf(256 * 1024); + SHA256_CTX hash; + SHA256_Init(&hash); + for (int64_t off = 0; off < size;) { + auto n = ::pread(fd, buf.data(), std::min(buf.size(), size - off), off); + if (n <= 0) { + fail(400, "cannot read local blob"); + return; + } + SHA256_Update(&hash, buf.data(), n); + off += n; + photon::thread_yield(); + } + unsigned char sum[SHA256_DIGEST_LENGTH]; + SHA256_Final(sum, &hash); + const char hex[] = "0123456789abcdef"; + std::string actual = "sha256:"; + for (auto b : sum) { + actual += hex[b >> 4]; + actual += hex[b & 15]; + } + if (actual != digest) { + fail(400, "local blob digest mismatch"); + return; + } + // Keep normal read flags: O_CACHE_ONLY would also disable subsequent + // remote cache-miss reads on a shared store. pwrite itself fills the cache. + std::unique_ptr cached( + service->global_fs.cached_fs->open(url.c_str(), O_RDONLY)); + if (!cached || cached->ftruncate(size) != 0) { + fail(503, "cannot open managed cache entry"); + return; + } + for (int64_t off = 0; off < size;) { + const auto count = std::min(buf.size(), size - off); + auto n = ::pread(fd, buf.data(), count, off); + if (n != count || cached->pwrite(buf.data(), count, off) != count) { + fail(503, "cache import incomplete; remote fallback remains available"); + return; + } + off += count; + photon::thread_yield(); + } + code = 200; + msg = ublkd_msg_ok(); +} diff --git a/src/ublk/cache_import.h b/src/ublk/cache_import.h new file mode 100644 index 00000000..1cb829cc --- /dev/null +++ b/src/ublk/cache_import.h @@ -0,0 +1,26 @@ +/* + Copyright The Overlaybd Authors + + Licensed under the Apache License, Version 2.0 (the "License"); + you may not use this file except in compliance with the License. + You may obtain a copy of the License at + + http://www.apache.org/licenses/LICENSE-2.0 + + Unless required by applicable law or agreed to in writing, software + distributed under the License is distributed on an "AS IS" BASIS, + WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + See the License for the specific language governing permissions and + limitations under the License. +*/ + +#pragma once + +#include + +class ImageService; + +// Import a finalized local registry blob through the daemon's managed file +// cache. The caller must keep the source bytes immutable until this returns. +void ublkd_handle_cache_import(ImageService *service, const std::string &body, int &code, + std::string &msg); diff --git a/src/ublk/test/CMakeLists.txt b/src/ublk/test/CMakeLists.txt index 10e5935e..6bef38ef 100644 --- a/src/ublk/test/CMakeLists.txt +++ b/src/ublk/test/CMakeLists.txt @@ -22,3 +22,21 @@ add_test( NAME ublk_dispatch_test COMMAND ${EXECUTABLE_OUTPUT_PATH}/ublk_dispatch_test ) + +# Exercises the real Photon file cache; no ublk device or privileged kernel +# operations are needed. Keep assertions enabled in Release test builds. +add_executable(ublkd_cache_import_test cache_import_test.cpp) +target_compile_options(ublkd_cache_import_test PRIVATE -UNDEBUG) +target_link_libraries(ublkd_cache_import_test + ublk_frontend_lib + photon_static + overlaybd_image_lib + libublksrv_static + ${CURL_LIBRARIES} + ${OPENSSL_SSL_LIBRARY} + ${OPENSSL_CRYPTO_LIBRARY} + ${AIO_LIBRARIES} +) +add_test(NAME ublkd_cache_import_test + COMMAND ${EXECUTABLE_OUTPUT_PATH}/ublkd_cache_import_test) +set_tests_properties(ublkd_cache_import_test PROPERTIES TIMEOUT 180) diff --git a/src/ublk/test/cache_import_test.cpp b/src/ublk/test/cache_import_test.cpp new file mode 100644 index 00000000..ae15c972 --- /dev/null +++ b/src/ublk/test/cache_import_test.cpp @@ -0,0 +1,202 @@ +/* + Copyright The Overlaybd Authors + + Licensed under the Apache License, Version 2.0 (the "License"); + you may not use this file except in compliance with the License. + You may obtain a copy of the License at + + http://www.apache.org/licenses/LICENSE-2.0 + + Unless required by applicable law or agreed to in writing, software + distributed under the License is distributed on an "AS IS" BASIS, + WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + See the License for the specific language governing permissions and + limitations under the License. +*/ + +// Integration test for the managed Photon cache: local import, offline hit, +// automatic capacity eviction, and source fallback. No KVM or ublk is needed. +#include +#include +#include +#include +#include +#include "../../image_service.h" +#include "../ublkd_protocol.h" +#include +#include +#include +#include +#include +#include +#include +#include +#include +#include "../cache_import.h" + +size_t cache_fn_trans_sha256(void *, std::string_view, char *, size_t); + +class CountingSource : public photon::fs::ForwardFS_Ownership { +public: + int opens = 0; + explicit CountingSource(const char *root) + : ForwardFS_Ownership(photon::fs::new_localfs_adaptor(root), true) { + } + photon::fs::IFile *open(const char *path, int flags, mode_t) override { + return open(path, flags); + } + photon::fs::IFile *open(const char *path, int flags) override { + ++opens; + std::string name(path); + name = name.substr(name.rfind('/') + 1); + return m_fs->open(name.c_str(), flags); + } +}; + +static std::string make_blob(const std::string &root, size_t size, unsigned char tag) { + std::vector block(256 * 1024, tag); + SHA256_CTX hash; + SHA256_Init(&hash); + std::string tmp = root + "/input"; + std::ofstream out(tmp, std::ios::binary); + for (size_t off = 0; off < size; off += block.size()) { + auto n = std::min(block.size(), size - off); + out.write((char *)block.data(), n); + SHA256_Update(&hash, block.data(), n); + } + out.close(); + unsigned char sum[32]; + SHA256_Final(sum, &hash); + std::string digest = "sha256:"; + for (auto c : sum) { + digest += "0123456789abcdef"[c >> 4]; + digest += "0123456789abcdef"[c & 15]; + } + assert(::rename(tmp.c_str(), (root + "/" + digest).c_str()) == 0); + return digest; +} + +static int import(ImageService &service, const std::string &path, const std::string &digest, + size_t size) { + std::string body = "{\"path\":\"" + path + "\",\"url\":\"https://registry.test/v2/repo/blobs/" + + digest + "\",\"digest\":\"" + digest + + "\",\"size\":" + std::to_string(size) + "}"; + int code = 0; + std::string msg; + ublkd_handle_cache_import(&service, body, code, msg); + if (code != 200) + std::cout << "import status=" << code << " " << msg << std::endl; + return code; +} + +static int remove_entry(const char *path, const struct stat *, int, struct FTW *) { + return ::remove(path); +} + +int main() { + assert(photon::init(photon::INIT_EVENT_EPOLL, photon::INIT_IO_DEFAULT) == 0); + char root[] = "/tmp/cache-import-test-XXXXXX"; + assert(mkdtemp(root)); + { + std::string source = std::string(root) + "/source", media = std::string(root) + "/cache"; + assert(::mkdir(source.c_str(), 0700) == 0); + assert(::mkdir(media.c_str(), 0700) == 0); + ImageService service; + std::string configPath = std::string(root) + "/config.json"; + std::ofstream(configPath) << "{\"cacheConfig\":{\"cacheType\":\"file\"}}"; + service.global_conf.ParseJSON(configPath); + auto src = new CountingSource(source.c_str()); + service.global_fs.srcfs = src; + service.global_fs.io_alloc = new IOAlloc; + auto cache = photon::fs::new_full_file_cached_fs( + src, photon::fs::new_localfs_adaptor(media.c_str()), 256 * 1024, 1, 10000, 0, + service.global_fs.io_alloc, 0, {nullptr, &cache_fn_trans_sha256}, 0); + assert(cache); + service.global_fs.cached_fs = cache; + const size_t small = 1024 * 1024 + 17; // exercise a non-block-aligned tail + auto first = make_blob(source, small, 42); + assert(import(service, source + "/" + first, first, small) == 200); + assert(src->opens == 0); + // Reject invalid requests before touching the managed cache. + int status = 0; + std::string message; + for (const auto &body : {"", "[]", "{}", "{\"path\":1}"}) { + ublkd_handle_cache_import(&service, body, status, message); + assert(status == 400); + } + for (const auto &fields : + {"\"path\":\"relative\",\"url\":\"https://registry.test/v2/repo/blobs/", + "\"path\":\"/tmp/blob\",\"url\":\"file:///blobs/"}) { + std::string body = + "{" + std::string(fields) + first + "\",\"digest\":\"" + first + "\",\"size\":1}"; + ublkd_handle_cache_import(&service, body, status, message); + assert(status == 400); + } + assert(import(service, source + "/" + first, first, 0) == 400); + assert(import(service, source + "/" + first, first, small + 1) == 400); + assert(import(service, source, first, small) == 400); + assert(import(service, source + "/missing", first, small) == 400); + assert(::symlink((source + "/" + first).c_str(), (source + "/link").c_str()) == 0); + assert(import(service, source + "/link", first, small) == 400); + assert(::unlink((source + "/link").c_str()) == 0); + assert(import(service, source + "/" + first, "sha256:bad", small) == 400); + std::ofstream(configPath) << "{\"cacheConfig\":{\"cacheType\":\"ocf\"}}"; + service.global_conf.ParseJSON(configPath); + assert(import(service, source + "/" + first, first, small) == 409); + std::ofstream(configPath) << "{\"cacheConfig\":{\"cacheType\":\"file\"}}"; + service.global_conf.ParseJSON(configPath); + assert(::unlink((source + "/" + first).c_str()) == 0); + auto url = "https://registry.test/v2/repo/blobs/" + first; + std::unique_ptr file(cache->open(url.c_str(), O_RDONLY)); + std::vector bytes(small); + assert(file && file->pread(bytes.data(), small, 0) == (ssize_t)small); + for (auto b : bytes) + assert(b == 42); + assert(src->opens == 0); + std::cout << "PASS imported blob readable after staging deletion, source opens=0" + << std::endl; + assert(make_blob(source, small, 42) == first); + // A bad digest must not poison an existing cache entry. + auto bad = make_blob(source, small, 43); + assert(import(service, source + "/" + bad, first, small) == 400); + assert(file->pread(bytes.data(), small, 0) == (ssize_t)small && bytes[0] == 42); + std::cout << "PASS digest mismatch rejected before cache writes" << std::endl; + // Drop the reader so normal LRU can reclaim this oldest entry. + file.reset(); + for (int n = 0; n < 12; ++n) { + auto digest = make_blob(source, 100 * 1024 * 1024, 60 + n); + auto code = import(service, source + "/" + digest, digest, 100 * 1024 * 1024); + assert(code == 200 || code == 503); + ::unlink((source + "/" + digest).c_str()); + photon::thread_usleep(20000); + } + struct stat st; + auto cachedPath = media + "/" + first; + bool evicted = ::stat(cachedPath.c_str(), &st) != 0 || st.st_blocks == 0; + assert(evicted); + uint64_t used = 0; + auto dir = ::opendir(media.c_str()); + assert(dir); + while (auto entry = ::readdir(dir)) { + if (entry->d_name[0] == '.') + continue; + struct stat entryStat; + assert(::stat((media + "/" + entry->d_name).c_str(), &entryStat) == 0); + used += entryStat.st_blocks * 512; + } + ::closedir(dir); + assert(used <= 1024ULL * 1024 * 1024); + std::cout << "PASS automatic capacity GC evicted imported blob, cache bytes=" << used + << std::endl; + file.reset(cache->open(url.c_str(), O_RDONLY)); + assert(file && file->pread(bytes.data(), small, 0) == (ssize_t)small); + for (auto b : bytes) + assert(b == 42); + assert(src->opens > 0); + std::cout << "PASS evicted blob refills from source, source opens=" << src->opens + << std::endl; + file.reset(); + } + assert(::nftw(root, remove_entry, 32, FTW_DEPTH | FTW_PHYS) == 0); + photon::fini(); +} diff --git a/src/ublk/ublkd_main.cpp b/src/ublk/ublkd_main.cpp index e58024e8..5c3f023c 100644 --- a/src/ublk/ublkd_main.cpp +++ b/src/ublk/ublkd_main.cpp @@ -23,6 +23,7 @@ #include "ublkd_protocol.h" #include "pool_placeholder.h" #include "blkdev_hygiene.h" +#include "cache_import.h" #include "../image_service.h" #include "../image_file.h" @@ -127,7 +128,9 @@ class UblkdServer : public photon::net::http::HTTPHandler { int code = 404; std::string msg = ublkd_msg_error("unknown endpoint: " + target); - if (target == "/v1/add" && req.verb() == photon::net::http::Verb::POST) { + if (target == "/v1/cache/import" && req.verb() == photon::net::http::Verb::POST) { + ublkd_handle_cache_import(service_, body, code, msg); + } else if (target == "/v1/add" && req.verb() == photon::net::http::Verb::POST) { handle_add(body, code, msg); } else if (target == "/v1/del" && req.verb() == photon::net::http::Verb::POST) { handle_del(body, code, msg); From 9bd218e04fa457abfaebfe6eec8f93b6d38a4ce5 Mon Sep 17 00:00:00 2001 From: Wenxiang Jin Date: Sat, 7 Nov 2026 10:30:18 +0800 Subject: [PATCH 2/4] Harden cache-owner local blob ingestion Move ingestion into ImageService and keep HTTP parsing in ublkd. Verify every reread chunk against the validated first pass before cache writes. Bound admission and proof memory, reject cache-bypassing read paths, and document synchronous best-effort semantics. Add deterministic overlap, mutation, truncation and write-failure tests; allow the cache integration target without the ublk frontend. --- docs/standalone-usage.md | 8 +- src/CMakeLists.txt | 1 + src/cache_import.cpp | 130 ++++++++++++++++++++++++++++ src/image_service.h | 13 +++ src/test/CMakeLists.txt | 22 +++++ src/ublk/cache_import.cpp | 91 +++---------------- src/ublk/cache_import.h | 3 +- src/ublk/test/CMakeLists.txt | 18 ---- src/ublk/test/cache_import_test.cpp | 128 ++++++++++++++++++++++++++- 9 files changed, 310 insertions(+), 104 deletions(-) create mode 100644 src/cache_import.cpp diff --git a/docs/standalone-usage.md b/docs/standalone-usage.md index c2cd0c5b..4b55df35 100644 --- a/docs/standalone-usage.md +++ b/docs/standalone-usage.md @@ -124,9 +124,13 @@ curl --unix-socket "$SOCK" -X POST -H 'Content-Type: application/json' \ http://d/v1/cache/import ``` -The source must be a regular file visible in the daemon's mount namespace; the final path component cannot be a symlink. Keep its contents immutable until the request completes. The daemon verifies the positive size and SHA-256 before writing any cache bytes. The URL must be an HTTP(S) registry blob URL ending in `/blobs/` and should match the URL used by subsequent image reads. This root-only local operation does not publish the blob or authenticate to the registry; publish durable content before importing it. +The source must be a regular file visible in the daemon's mount namespace; the final path component cannot be a symlink. Keep its contents immutable until the request completes. The daemon verifies the positive size and SHA-256 before writing any cache bytes, then verifies each reread chunk against the first pass before publishing it. In-place mutation cannot publish unchecked bytes; an error may leave an already verified prefix in cache. The URL must be an HTTP(S) registry blob URL ending in `/blobs/` and should match the URL used by subsequent image reads. This root-only local operation does not publish the blob or authenticate to the registry; publish durable content before importing it. -The endpoint returns `200` on completion, `400` for invalid requests or local content, `409` when the managed file cache is unavailable or another cache type is configured, and `503` if cache population cannot complete. Import is an optional optimization: failures, partial imports, and later eviction retain the normal source-read path. Producers may remove their staging files after the request finishes. Cache imports use the same capacity, LRU eviction, allocation accounting, and free-space protection as downloaded blobs; they do not pin entries or create a separate retained copy. Existing active device and mount lifecycles are unchanged. +The endpoint returns `200` on completion, `400` for invalid requests or local content, `409` when the managed file cache is unavailable or another cache type is configured, `413` for blobs above 64 GiB, `429` while another import is active, and `503` if cache population cannot complete. A P2P read path that bypasses the managed cache at admission also returns `409`; later read-path changes are not serialized with imports. + +Imports are synchronous, with one active import per ImageService and no waiting queue. Disconnecting the HTTP client does not cancel an admitted import; wait for completion or retry with backoff on `429`. Local file reads are synchronous and yield between 256 KiB chunks, so slow local storage can still delay daemon work. Integrity proofs use at most 8 MiB plus a 256 KiB data buffer per active import. A `200` means all writes were accepted, not that bytes are pinned, durable across crashes, or guaranteed resident after return. + +Import is an optional optimization: failures, partial imports, and later eviction retain the normal source-read path. Producers may remove their staging files after the request finishes. Cache imports use the same capacity, LRU eviction, allocation accounting, and free-space protection as downloaded blobs; they do not pin entries or create a separate retained copy. The ingestion API belongs to ImageService and is available independently of the ublk HTTP adapter. Registry publication, readiness checks, retries, staging cleanup and Fast Sandbox orchestration remain the producer's responsibility. Existing active device and mount lifecycles are unchanged. ### Notes diff --git a/src/CMakeLists.txt b/src/CMakeLists.txt index 552a3481..4a68a9a5 100644 --- a/src/CMakeLists.txt +++ b/src/CMakeLists.txt @@ -10,6 +10,7 @@ add_subdirectory(overlaybd) add_library(overlaybd_image_lib image_file.cpp image_service.cpp + cache_import.cpp switch_file.cpp bk_download.cpp prefetch.cpp diff --git a/src/cache_import.cpp b/src/cache_import.cpp new file mode 100644 index 00000000..43f2bd57 --- /dev/null +++ b/src/cache_import.cpp @@ -0,0 +1,130 @@ +/* + Copyright The Overlaybd Authors + + Licensed under the Apache License, Version 2.0 (the "License"); + you may not use this file except in compliance with the License. + You may obtain a copy of the License at + + http://www.apache.org/licenses/LICENSE-2.0 + + Unless required by applicable law or agreed to in writing, software + distributed under the License is distributed on an "AS IS" BASIS, + WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + See the License for the specific language governing permissions and + limitations under the License. +*/ + +#include "image_service.h" +#include +#include +#include +#include +#include +#include +#include +#include +#include + +namespace { +// POSIX reads may be short or interrupted, including reads of regular files. +ssize_t read_chunk(int fd, unsigned char *buffer, size_t count, off_t offset) { + size_t done = 0; + while (done < count) { + auto n = ::pread(fd, buffer + done, count - done, offset + done); + if (n < 0 && errno == EINTR) continue; + if (n <= 0) return -1; + done += n; + } + return done; +} +} // namespace + +CacheImportResult ImageService::import_local_blob(const std::string &path, + const std::string &url, const std::string &digest, int64_t size) { + const std::string suffix = "/blobs/" + digest; + if (size <= 0 || digest.size() != 71 || digest.substr(0, 7) != "sha256:" || + digest.find_first_not_of("0123456789abcdef", 7) != std::string::npos || path.empty() || + path[0] != '/' || path.find('\0') != std::string::npos || + url.find('\0') != std::string::npos || + (url.substr(0, 7) != "http://" && url.substr(0, 8) != "https://") || + url.size() <= suffix.size() || + url.compare(url.size() - suffix.size(), suffix.size(), suffix) != 0) { + return {CacheImportStatus::invalid, "invalid content-addressed cache import"}; + } + // Bound proof memory to 8 MiB and reject overload instead of building an + // unbounded queue of expensive synchronous imports. + if (size > 64LL * 1024 * 1024 * 1024) + return {CacheImportStatus::too_large, "local blob exceeds 64 GiB import limit"}; + if (m_cache_import_active.test_and_set()) + return {CacheImportStatus::busy, "another local blob import is active; retry later"}; + DEFER(m_cache_import_active.clear()); + if (global_conf.cacheConfig().cacheType() != "file" || !global_fs.cached_fs) + return {CacheImportStatus::unsupported, "local import requires the managed file cache"}; + if (global_fs.remote_fs && global_fs.remote_fs != global_fs.cached_fs) + return {CacheImportStatus::unsupported, "active remote read path bypasses the managed cache"}; + int fd = ::open(path.c_str(), O_RDONLY | O_CLOEXEC | O_NOFOLLOW | O_NONBLOCK); + if (fd < 0) { + return {CacheImportStatus::invalid, "cannot open local blob"}; + } + DEFER(::close(fd)); + struct stat st; + if (::fstat(fd, &st) != 0 || !S_ISREG(st.st_mode) || st.st_size != size) { + return {CacheImportStatus::invalid, "local blob size or type mismatch"}; + } + // Validate the whole file before exposing any bytes under its digest. + std::vector buf(256 * 1024); + std::vector> chunks; + chunks.reserve((size + buf.size() - 1) / buf.size()); + SHA256_CTX hash; + SHA256_Init(&hash); + for (int64_t off = 0; off < size;) { + const auto count = std::min(buf.size(), size - off); + auto n = read_chunk(fd, buf.data(), count, off); + if (n != count) { + return {CacheImportStatus::invalid, "cannot read local blob"}; + } + SHA256_Update(&hash, buf.data(), n); + std::array chunk; + SHA256(buf.data(), n, chunk.data()); + chunks.push_back(chunk); + off += n; + photon::thread_yield(); + } + unsigned char sum[SHA256_DIGEST_LENGTH]; + SHA256_Final(sum, &hash); + const char hex[] = "0123456789abcdef"; + std::string actual = "sha256:"; + for (auto b : sum) { + actual += hex[b >> 4]; + actual += hex[b & 15]; + } + if (actual != digest) { + return {CacheImportStatus::invalid, "local blob digest mismatch"}; + } + // Keep normal read flags: O_CACHE_ONLY would also disable subsequent + // remote cache-miss reads on a shared store. pwrite itself fills the cache. + std::unique_ptr cached( + global_fs.cached_fs->open(url.c_str(), O_RDONLY)); + if (!cached || cached->ftruncate(size) != 0) { + return {CacheImportStatus::unavailable, "cannot open managed cache entry"}; + } + for (int64_t off = 0; off < size;) { + const auto count = std::min(buf.size(), size - off); + auto n = read_chunk(fd, buf.data(), count, off); + if (n != count) { + return {CacheImportStatus::invalid, "local blob changed during import"}; + } + std::array chunk; + SHA256(buf.data(), count, chunk.data()); + if (chunk != chunks[off / buf.size()]) { + return {CacheImportStatus::invalid, "local blob changed during import"}; + } + // Only these verified bytes are published, never an unchecked reread. + if (cached->pwrite(buf.data(), count, off) != count) { + return {CacheImportStatus::unavailable, "cache import incomplete; remote fallback remains available"}; + } + off += count; + photon::thread_yield(); + } + return {CacheImportStatus::ok, ""}; +} diff --git a/src/image_service.h b/src/image_service.h index deeb1424..e2a2ee97 100644 --- a/src/image_service.h +++ b/src/image_service.h @@ -17,6 +17,8 @@ #pragma once #include +#include +#include #include "config.h" #include "exporter_server.h" #include "overlaybd/gzip_cache/cached_fs.h" @@ -51,11 +53,21 @@ struct ImageAuthResponse : public ConfigUtils::Config { struct ImageFile; struct ApiServer; +enum class CacheImportStatus { ok, invalid, unsupported, too_large, busy, unavailable }; +struct CacheImportResult { + CacheImportStatus status; + const char *message; +}; + class ImageService { public: ImageService(const char *config_path = nullptr); ~ImageService(); int init(); + // Synchronous best-effort cache seeding. No publication, staging ownership, + // or HTTP policy lives here. Partial verified fills remain safe to read. + CacheImportResult import_local_blob(const std::string &path, const std::string &url, + const std::string &digest, int64_t size); ImageFile *create_image_file(const char *image_config_path, const std::string &dev_id); // bool enable_acceleration(GlobalFs *global_fs, ImageConfigNS::P2PConfig conf); bool enable_acceleration(); @@ -71,6 +83,7 @@ class ImageService { ApiServer *api_server = nullptr; private: + std::atomic_flag m_cache_import_active = ATOMIC_FLAG_INIT; int read_global_config_and_set(); std::pair reload_auth(const char *remote_path); void set_result_file(std::string &filename, std::string &data); diff --git a/src/test/CMakeLists.txt b/src/test/CMakeLists.txt index 7fa107e5..cedcfd7e 100644 --- a/src/test/CMakeLists.txt +++ b/src/test/CMakeLists.txt @@ -74,3 +74,25 @@ if (NOT ORIGIN_EXT2FS) COMMAND ${EXECUTABLE_OUTPUT_PATH}/resize_test ) endif() + +# Exercises the real Photon file cache; no ublk device or privileged kernel +# operations are needed. Keep assertions enabled in Release test builds. +# Keep this available with BUILD_UBLK_FRONTEND=OFF: neither ingestion nor its +# HTTP parsing tests depend on libublksrv or a ublk-capable kernel. +add_executable(ublkd_cache_import_test + ../ublk/test/cache_import_test.cpp + ../ublk/cache_import.cpp + ../ublk/ublkd_protocol.cpp +) +target_compile_options(ublkd_cache_import_test PRIVATE -UNDEBUG) +target_link_libraries(ublkd_cache_import_test + photon_static + overlaybd_image_lib + ${CURL_LIBRARIES} + ${OPENSSL_SSL_LIBRARY} + ${OPENSSL_CRYPTO_LIBRARY} + ${AIO_LIBRARIES} +) +add_test(NAME ublkd_cache_import_test + COMMAND ${EXECUTABLE_OUTPUT_PATH}/ublkd_cache_import_test) +set_tests_properties(ublkd_cache_import_test PROPERTIES TIMEOUT 180) diff --git a/src/ublk/cache_import.cpp b/src/ublk/cache_import.cpp index b514de1c..b7169379 100644 --- a/src/ublk/cache_import.cpp +++ b/src/ublk/cache_import.cpp @@ -17,16 +17,7 @@ #include "cache_import.h" #include "ublkd_protocol.h" #include "../image_service.h" -#include -#include -#include - -#include -#include -#include -#include #include -#include void ublkd_handle_cache_import(ImageService *service, const std::string &body, int &code, std::string &msg) { @@ -47,76 +38,14 @@ void ublkd_handle_cache_import(ImageService *service, const std::string &body, i std::string url(doc["url"].GetString(), doc["url"].GetStringLength()); std::string digest(doc["digest"].GetString(), doc["digest"].GetStringLength()); const int64_t size = doc["size"].GetInt64(); - const std::string suffix = "/blobs/" + digest; - if (digest.size() != 71 || digest.substr(0, 7) != "sha256:" || - digest.find_first_not_of("0123456789abcdef", 7) != std::string::npos || path.empty() || - path[0] != '/' || path.find('\0') != std::string::npos || - url.find('\0') != std::string::npos || - (url.substr(0, 7) != "http://" && url.substr(0, 8) != "https://") || - url.size() <= suffix.size() || - url.compare(url.size() - suffix.size(), suffix.size(), suffix) != 0) { - fail(400, "invalid content-addressed cache import"); - return; - } - if (service->global_conf.cacheConfig().cacheType() != "file" || !service->global_fs.cached_fs) { - fail(409, "local import requires the managed file cache"); - return; - } - int fd = ::open(path.c_str(), O_RDONLY | O_CLOEXEC | O_NOFOLLOW | O_NONBLOCK); - if (fd < 0) { - fail(400, "cannot open local blob"); - return; - } - DEFER(::close(fd)); - struct stat st; - if (::fstat(fd, &st) != 0 || !S_ISREG(st.st_mode) || st.st_size != size) { - fail(400, "local blob size or type mismatch"); - return; - } - // Validate the whole file before exposing any bytes under its digest. - std::vector buf(256 * 1024); - SHA256_CTX hash; - SHA256_Init(&hash); - for (int64_t off = 0; off < size;) { - auto n = ::pread(fd, buf.data(), std::min(buf.size(), size - off), off); - if (n <= 0) { - fail(400, "cannot read local blob"); - return; - } - SHA256_Update(&hash, buf.data(), n); - off += n; - photon::thread_yield(); - } - unsigned char sum[SHA256_DIGEST_LENGTH]; - SHA256_Final(sum, &hash); - const char hex[] = "0123456789abcdef"; - std::string actual = "sha256:"; - for (auto b : sum) { - actual += hex[b >> 4]; - actual += hex[b & 15]; - } - if (actual != digest) { - fail(400, "local blob digest mismatch"); - return; - } - // Keep normal read flags: O_CACHE_ONLY would also disable subsequent - // remote cache-miss reads on a shared store. pwrite itself fills the cache. - std::unique_ptr cached( - service->global_fs.cached_fs->open(url.c_str(), O_RDONLY)); - if (!cached || cached->ftruncate(size) != 0) { - fail(503, "cannot open managed cache entry"); - return; - } - for (int64_t off = 0; off < size;) { - const auto count = std::min(buf.size(), size - off); - auto n = ::pread(fd, buf.data(), count, off); - if (n != count || cached->pwrite(buf.data(), count, off) != count) { - fail(503, "cache import incomplete; remote fallback remains available"); - return; - } - off += count; - photon::thread_yield(); - } - code = 200; - msg = ublkd_msg_ok(); + auto result = service->import_local_blob(path, url, digest, size); + switch (result.status) { + case CacheImportStatus::ok: code = 200; break; + case CacheImportStatus::invalid: code = 400; break; + case CacheImportStatus::unsupported: code = 409; break; + case CacheImportStatus::too_large: code = 413; break; + case CacheImportStatus::busy: code = 429; break; + case CacheImportStatus::unavailable: code = 503; break; + } + msg = code == 200 ? ublkd_msg_ok() : ublkd_msg_error(result.message); } diff --git a/src/ublk/cache_import.h b/src/ublk/cache_import.h index 1cb829cc..0e8ec4e0 100644 --- a/src/ublk/cache_import.h +++ b/src/ublk/cache_import.h @@ -20,7 +20,6 @@ class ImageService; -// Import a finalized local registry blob through the daemon's managed file -// cache. The caller must keep the source bytes immutable until this returns. +// HTTP parsing and status mapping for ImageService::import_local_blob. void ublkd_handle_cache_import(ImageService *service, const std::string &body, int &code, std::string &msg); diff --git a/src/ublk/test/CMakeLists.txt b/src/ublk/test/CMakeLists.txt index 6bef38ef..10e5935e 100644 --- a/src/ublk/test/CMakeLists.txt +++ b/src/ublk/test/CMakeLists.txt @@ -22,21 +22,3 @@ add_test( NAME ublk_dispatch_test COMMAND ${EXECUTABLE_OUTPUT_PATH}/ublk_dispatch_test ) - -# Exercises the real Photon file cache; no ublk device or privileged kernel -# operations are needed. Keep assertions enabled in Release test builds. -add_executable(ublkd_cache_import_test cache_import_test.cpp) -target_compile_options(ublkd_cache_import_test PRIVATE -UNDEBUG) -target_link_libraries(ublkd_cache_import_test - ublk_frontend_lib - photon_static - overlaybd_image_lib - libublksrv_static - ${CURL_LIBRARIES} - ${OPENSSL_SSL_LIBRARY} - ${OPENSSL_CRYPTO_LIBRARY} - ${AIO_LIBRARIES} -) -add_test(NAME ublkd_cache_import_test - COMMAND ${EXECUTABLE_OUTPUT_PATH}/ublkd_cache_import_test) -set_tests_properties(ublkd_cache_import_test PROPERTIES TIMEOUT 180) diff --git a/src/ublk/test/cache_import_test.cpp b/src/ublk/test/cache_import_test.cpp index ae15c972..850ba427 100644 --- a/src/ublk/test/cache_import_test.cpp +++ b/src/ublk/test/cache_import_test.cpp @@ -19,6 +19,8 @@ #include #include #include +#include +#include #include #include #include "../../image_service.h" @@ -27,6 +29,7 @@ #include #include #include +#include #include #include #include @@ -53,6 +56,41 @@ class CountingSource : public photon::fs::ForwardFS_Ownership { } }; +// Deterministic fault boundaries: open happens after full digest validation; +// each pwrite happens after the corresponding source chunk is revalidated. +class ImportFaultFile : public photon::fs::ForwardFile_Ownership { +public: + std::function after_first; + bool fail_second; + int writes = 0; + ImportFaultFile(photon::fs::IFile *file, std::function hook, bool fail) + : ForwardFile_Ownership(file, true), after_first(hook), fail_second(fail) {} + ssize_t pwrite(const void *buf, size_t count, off_t offset) override { + if (++writes == 2 && fail_second) { + errno = EIO; + return -1; + } + auto n = m_file->pwrite(buf, count, offset); + if (writes == 1 && after_first) after_first(); + return n; + } +}; + +class ImportFaultFS : public photon::fs::ForwardFS_Ownership { +public: + std::function on_open, after_first; + bool fail_second = false; + explicit ImportFaultFS(photon::fs::IFileSystem *fs) : ForwardFS_Ownership(fs, false) {} + photon::fs::IFile *open(const char *path, int flags, mode_t) override { + return open(path, flags); + } + photon::fs::IFile *open(const char *path, int flags) override { + if (on_open) on_open(); + auto file = m_fs->open(path, flags); + return file ? new ImportFaultFile(file, after_first, fail_second) : nullptr; + } +}; + static std::string make_blob(const std::string &root, size_t size, unsigned char tag) { std::vector block(256 * 1024, tag); SHA256_CTX hash; @@ -161,6 +199,94 @@ int main() { assert(import(service, source + "/" + bad, first, small) == 400); assert(file->pread(bytes.data(), small, 0) == (ssize_t)small && bytes[0] == 42); std::cout << "PASS digest mismatch rejected before cache writes" << std::endl; + // P2P actually using srcfs bypasses these cached bytes: never report a + // successful warmup for that read path. Oversized requests are bounded. + service.global_fs.remote_fs = src; + assert(import(service, source + "/" + first, first, small) == 409); + service.global_fs.remote_fs = nullptr; + assert(import(service, source + "/" + first, first, + 64ULL * 1024 * 1024 * 1024 + 1) == 413); + { + ImportFaultFS fault(cache); + service.global_fs.cached_fs = &fault; + const auto staged = source + "/staged"; + auto prepare = [&](unsigned char tag) { + auto digest = make_blob(source, small, tag); + std::ifstream in(source + "/" + digest, std::ios::binary); + std::ofstream out(staged, std::ios::binary); + out << in.rdbuf(); + return digest; + }; + auto overwrite = [&](off_t offset) { + int fd = ::open(staged.c_str(), O_WRONLY); + assert(fd >= 0); + unsigned char changed = 255; + assert(::pwrite(fd, &changed, 1, offset) == 1); + assert(::close(fd) == 0); + }; + auto read_back = [&](const std::string &digest, unsigned char tag, + bool has_prefix = false) { + auto name = "https://registry.test/v2/repo/blobs/" + digest; + std::unique_ptr reader(cache->open(name.c_str(), O_RDONLY)); + std::vector data(small); + assert(reader); + if (has_prefix) { + int opens = src->opens; + assert(reader->pread(data.data(), 256 * 1024, 0) == 256 * 1024); + for (size_t i = 0; i < 256 * 1024; ++i) assert(data[i] == tag); + assert(src->opens == opens); + } + assert(reader->pread(data.data(), small, 0) == (ssize_t)small); + for (auto b : data) assert(b == tag); + }; + // An overlapping request is rejected without waiting. The outer + // request and a subsequent retry must both remain usable. + auto concurrent = prepare(80); + fault.on_open = [&] { + auto contender = photon::thread_create11([&] { + assert(import(service, staged, concurrent, small) == 429); + }); + assert(contender); + photon::thread_join(photon::thread_enable_join(contender)); + }; + assert(import(service, staged, concurrent, small) == 200); + fault.on_open = nullptr; + assert(import(service, staged, concurrent, small) == 200); + read_back(concurrent, 80); + + // Mutation exactly between the two passes must publish no corrupt + // bytes. A separately preserved remote blob supplies cache misses. + auto changed = prepare(81); + fault.on_open = [&] { overwrite(0); }; + assert(import(service, staged, changed, small) == 400); + fault.on_open = nullptr; + read_back(changed, 81); + + // Mutation after a verified prefix has been filled must leave that + // prefix safe and allow the untouched source to refill the suffix. + auto partial = prepare(82); + fault.after_first = [&] { overwrite(256 * 1024); }; + assert(import(service, staged, partial, small) == 400); + fault.after_first = nullptr; + read_back(partial, 82, true); + + auto truncated = prepare(83); + fault.after_first = [&] { assert(::truncate(staged.c_str(), 0) == 0); }; + assert(import(service, staged, truncated, small) == 400); + fault.after_first = nullptr; + read_back(truncated, 83, true); + + auto interrupted = prepare(84); + fault.fail_second = true; + assert(import(service, staged, interrupted, small) == 503); + fault.fail_second = false; + read_back(interrupted, 84, true); + assert(import(service, staged, interrupted, small) == 200); + service.global_fs.cached_fs = cache; + assert(::unlink(staged.c_str()) == 0); + } + std::cout << "PASS admission, source mutation, truncation and write failure fallback" + << std::endl; // Drop the reader so normal LRU can reclaim this oldest entry. file.reset(); for (int n = 0; n < 12; ++n) { @@ -181,7 +307,7 @@ int main() { if (entry->d_name[0] == '.') continue; struct stat entryStat; - assert(::stat((media + "/" + entry->d_name).c_str(), &entryStat) == 0); + assert(::stat((media + "/" + std::string(entry->d_name)).c_str(), &entryStat) == 0); used += entryStat.st_blocks * 512; } ::closedir(dir); From c2b2da919938e65a4ad343bd6a64a84aa24012b5 Mon Sep 17 00:00:00 2001 From: Wenxiang Jin Date: Mon, 9 Nov 2026 10:36:26 +0800 Subject: [PATCH 3/4] Support cache-first P2P blob imports Add opt-in p2pConfig.cacheFirst so reads consult the managed file cache before fetching missing ranges through the accelerator. Preserve the default bypass mode and reject imports under that policy before image creation, when the remote read path has not yet been selected. Resolve the effective cache type consistently for initialization and import admission, including legacy and default configurations. Test real RegistryFS v1/v2 HTTP reads, imported hits, partial and full eviction/refill, unavailable accelerators, bypass compatibility, and legacy cache admission. Document configuration and storage tradeoffs. Signed-off-by: Wenxiang Jin --- README.md | 1 + docs/standalone-usage.md | 21 ++- src/cache_import.cpp | 7 +- src/config.h | 6 + src/image_service.cpp | 10 +- src/test/CMakeLists.txt | 9 ++ src/test/p2p_cache_test.cpp | 221 ++++++++++++++++++++++++++++ src/ublk/test/cache_import_test.cpp | 16 ++ 8 files changed, 285 insertions(+), 6 deletions(-) create mode 100644 src/test/p2p_cache_test.cpp diff --git a/README.md b/README.md index dc49a7bb..a9241a6c 100644 --- a/README.md +++ b/README.md @@ -228,6 +228,7 @@ Default configure file `overlaybd.json` is installed to `/etc/overlaybd/`. | download.maxMBps | The speed limit in MB/s for a downloading task. | | download.blockSize | The download block size from source, in byte. `262144` is default (256 KB). | | p2pConfig.enable | Whether p2p proxy is enabled or not. | +| p2pConfig.cacheFirst | Serve reads from the managed file cache before fetching misses through P2P. Defaults to `false` (the existing P2P cache bypass); requires the effective cache type to be `file`. See [local cache first with P2P](docs/standalone-usage.md#local-cache-first-with-p2p). | | p2pConfig.address | The proxy for p2p download, the format is `localhost:/`, depending on dadip2p.yaml | | exporterConfig.enable | whether or not create a server to show Prometheus metrics. | | exporterConfig.uriPrefix | URI prefix for export metrics. | diff --git a/docs/standalone-usage.md b/docs/standalone-usage.md index 4b55df35..71c811c6 100644 --- a/docs/standalone-usage.md +++ b/docs/standalone-usage.md @@ -126,12 +126,31 @@ curl --unix-socket "$SOCK" -X POST -H 'Content-Type: application/json' \ The source must be a regular file visible in the daemon's mount namespace; the final path component cannot be a symlink. Keep its contents immutable until the request completes. The daemon verifies the positive size and SHA-256 before writing any cache bytes, then verifies each reread chunk against the first pass before publishing it. In-place mutation cannot publish unchecked bytes; an error may leave an already verified prefix in cache. The URL must be an HTTP(S) registry blob URL ending in `/blobs/` and should match the URL used by subsequent image reads. This root-only local operation does not publish the blob or authenticate to the registry; publish durable content before importing it. -The endpoint returns `200` on completion, `400` for invalid requests or local content, `409` when the managed file cache is unavailable or another cache type is configured, `413` for blobs above 64 GiB, `429` while another import is active, and `503` if cache population cannot complete. A P2P read path that bypasses the managed cache at admission also returns `409`; later read-path changes are not serialized with imports. +The endpoint returns `200` on completion, `400` for invalid requests or local content, `409` when the managed file cache is unavailable or another cache type is configured, `413` for blobs above 64 GiB, `429` while another import is active, and `503` if cache population cannot complete. Both `cacheConfig.cacheType: "file"` and the legacy top-level `cacheType: "file"` are supported; an omitted or empty nested type falls back to the top-level type, whose default is `file`. Imports are synchronous, with one active import per ImageService and no waiting queue. Disconnecting the HTTP client does not cancel an admitted import; wait for completion or retry with backoff on `429`. Local file reads are synchronous and yield between 256 KiB chunks, so slow local storage can still delay daemon work. Integrity proofs use at most 8 MiB plus a 256 KiB data buffer per active import. A `200` means all writes were accepted, not that bytes are pinned, durable across crashes, or guaranteed resident after return. Import is an optional optimization: failures, partial imports, and later eviction retain the normal source-read path. Producers may remove their staging files after the request finishes. Cache imports use the same capacity, LRU eviction, allocation accounting, and free-space protection as downloaded blobs; they do not pin entries or create a separate retained copy. The ingestion API belongs to ImageService and is available independently of the ublk HTTP adapter. Registry publication, readiness checks, retries, staging cleanup and Fast Sandbox orchestration remain the producer's responsibility. Existing active device and mount lifecycles are unchanged. +#### Local cache first with P2P + +To use imported blobs while P2P is enabled, opt into `p2pConfig.cacheFirst` in the daemon's service configuration: + +```json +{ + "cacheConfig": {"cacheType": "file"}, + "p2pConfig": { + "enable": true, + "address": "http://127.0.0.1:9731/accelerator", + "cacheFirst": true + } +} +``` + +In this mode, image reads always enter the managed file cache. Cached ranges are served locally; missing or evicted ranges are fetched through RegistryFS using the selected P2P accelerator and refill the same cache. Imports may run before the first image is created. Selecting or rechecking acceleration during image creation does not bypass the cache. If the accelerator availability check fails, RegistryFS uses the registry directly while keeping the local cache in front; this does not add per-request failover for accelerator failures after selection. + +`cacheFirst` defaults to `false`, preserving the existing P2P cache-bypass behavior, and requires the effective cache type to be `file`. When P2P is enabled without `cacheFirst`, imports return `409` even before the first image selects a read path: a later successful P2P probe could otherwise bypass the imported data. Set `cacheFirst: true` to combine P2P with imports, or disable P2P to use the ordinary file-cache path. Cache-first mode also caches data downloaded through P2P, so a P2P service with its own disk cache may retain a second local copy. Existing cache capacity and eviction policies still apply. + ### Notes * Always unmount filesystems on `/dev/ublkbN` before `del` (or before stopping diff --git a/src/cache_import.cpp b/src/cache_import.cpp index 43f2bd57..394fbca8 100644 --- a/src/cache_import.cpp +++ b/src/cache_import.cpp @@ -58,8 +58,13 @@ CacheImportResult ImageService::import_local_blob(const std::string &path, if (m_cache_import_active.test_and_set()) return {CacheImportStatus::busy, "another local blob import is active; retry later"}; DEFER(m_cache_import_active.clear()); - if (global_conf.cacheConfig().cacheType() != "file" || !global_fs.cached_fs) + if (global_conf.effectiveCacheType() != "file" || !global_fs.cached_fs) return {CacheImportStatus::unsupported, "local import requires the managed file cache"}; + // Decide from policy even before the first image selects remote_fs. A + // successful later P2P probe must not silently bypass an admitted warmup. + auto p2p = global_conf.p2pConfig(); + if (p2p.enable() && !p2p.cacheFirst()) + return {CacheImportStatus::unsupported, "local import with P2P requires p2pConfig.cacheFirst"}; if (global_fs.remote_fs && global_fs.remote_fs != global_fs.cached_fs) return {CacheImportStatus::unsupported, "active remote read path bypasses the managed cache"}; int fd = ::open(path.c_str(), O_RDONLY | O_CLOEXEC | O_NOFOLLOW | O_NONBLOCK); diff --git a/src/config.h b/src/config.h index f76ed8de..5b6e33b2 100644 --- a/src/config.h +++ b/src/config.h @@ -116,6 +116,7 @@ struct P2PConfig : public ConfigUtils::Config { APPCFG_CLASS APPCFG_PARA(enable, bool, false); + APPCFG_PARA(cacheFirst, bool, false); APPCFG_PARA(address, std::string, "http://localhost:9731/accelerator"); }; @@ -211,6 +212,11 @@ struct GlobalConfig : public ConfigUtils::Config { APPCFG_PARA(certConfig, CertConfig); APPCFG_PARA(userAgent, std::string, OVERLAYBD_VERSION); APPCFG_PARA(serviceConfig, ServiceConfig); + + std::string effectiveCacheType() { + auto type = cacheConfig().cacheType(); + return type.empty() ? cacheType() : type; + } }; struct AuthConfig : public ConfigUtils::Config { diff --git a/src/image_service.cpp b/src/image_service.cpp index 12dd0799..74b618bc 100644 --- a/src/image_service.cpp +++ b/src/image_service.cpp @@ -406,14 +406,12 @@ int ImageService::init() { return -1; } - std::string cache_type, cache_dir; + std::string cache_type = global_conf.effectiveCacheType(), cache_dir; uint32_t cache_size_GB, refill_size, block_size; if (global_conf.cacheConfig().cacheType().empty()) { - cache_type = global_conf.cacheType(); cache_dir = global_conf.registryCacheDir(); cache_size_GB = global_conf.registryCacheSizeGB(); } else { - cache_type = global_conf.cacheConfig().cacheType(); cache_dir = global_conf.cacheConfig().cacheDir(); cache_size_GB = global_conf.cacheConfig().cacheSizeGB(); } @@ -423,6 +421,9 @@ int ImageService::init() { if (cache_type != "file" && cache_type != "ocf" && cache_type != "download") { LOG_ERROR_RETURN(0, -1, "unknown cache type: `", cache_type); } + if (global_conf.p2pConfig().cacheFirst() && cache_type != "file") { + LOG_ERROR_RETURN(0, -1, "p2pConfig.cacheFirst requires the file cache"); + } LOG_INFO("cache config: ", VALUE(cache_type), VALUE(cache_dir), VALUE(cache_size_GB), VALUE(refill_size)); @@ -559,7 +560,8 @@ bool ImageService::enable_acceleration() { auto conf = global_conf.p2pConfig(); if (conf.enable() && check_accelerate_url(conf.address())) { ((RegistryFS*)global_fs.underlay_registryfs)->setAccelerateAddress(conf.address().c_str()); - global_fs.remote_fs = global_fs.srcfs; + // Cache misses still use srcfs, whose RegistryFS now targets P2P. + global_fs.remote_fs = conf.cacheFirst() ? global_fs.cached_fs : global_fs.srcfs; return true; } else { ((RegistryFS*)(global_fs.underlay_registryfs))->setAccelerateAddress(); diff --git a/src/test/CMakeLists.txt b/src/test/CMakeLists.txt index cedcfd7e..5342b405 100644 --- a/src/test/CMakeLists.txt +++ b/src/test/CMakeLists.txt @@ -96,3 +96,12 @@ target_link_libraries(ublkd_cache_import_test add_test(NAME ublkd_cache_import_test COMMAND ${EXECUTABLE_OUTPUT_PATH}/ublkd_cache_import_test) set_tests_properties(ublkd_cache_import_test PROPERTIES TIMEOUT 180) + +# Exercises actual RegistryFS requests through a loopback P2P accelerator. +add_executable(p2p_cache_test p2p_cache_test.cpp) +target_compile_options(p2p_cache_test PRIVATE -UNDEBUG) +target_link_libraries(p2p_cache_test + photon_static overlaybd_image_lib ${CURL_LIBRARIES} + ${OPENSSL_SSL_LIBRARY} ${OPENSSL_CRYPTO_LIBRARY} ${AIO_LIBRARIES}) +add_test(NAME p2p_cache_test COMMAND ${EXECUTABLE_OUTPUT_PATH}/p2p_cache_test) +set_tests_properties(p2p_cache_test PROPERTIES TIMEOUT 120) diff --git a/src/test/p2p_cache_test.cpp b/src/test/p2p_cache_test.cpp new file mode 100644 index 00000000..71039f32 --- /dev/null +++ b/src/test/p2p_cache_test.cpp @@ -0,0 +1,221 @@ +/* + Copyright The Overlaybd Authors + + Licensed under the Apache License, Version 2.0 (the "License"); + you may not use this file except in compliance with the License. + You may obtain a copy of the License at + + http://www.apache.org/licenses/LICENSE-2.0 + + Unless required by applicable law or agreed to in writing, software + distributed under the License is distributed on an "AS IS" BASIS, + WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + See the License for the specific language governing permissions and + limitations under the License. +*/ + +// Real ImageService + Photon file cache + loopback HTTP origin/accelerator. +// No external registry, ublk device, or privileged kernel operation is needed. +#include "../image_service.h" +#include +#include +#include +#include +#include +#include +#include +#include +#include +#include +#include +#include +#include + +using namespace photon::net; +using namespace photon::net::http; + +class BlobServer { +public: + const std::string data = std::string(768 * 1024 + 17, 'x'); + std::string digest, url, last_target; + int requests = 0; + ISocketServer *socket = new_tcp_socket_server(); + HTTPServer *http = new_http_server(); + + explicit BlobServer(bool listen = true) { + unsigned char hash[SHA256_DIGEST_LENGTH]; + SHA256((const unsigned char *)data.data(), data.size(), hash); + digest = "sha256:"; + for (auto byte : hash) { + digest += "0123456789abcdef"[byte >> 4]; + digest += "0123456789abcdef"[byte & 15]; + } + assert(socket && http); + assert(socket->bind(0, IPAddr("127.0.0.1")) == 0); + url = "http://127.0.0.1:" + std::to_string(socket->getsockname().port); + http->add_handler({this, &BlobServer::handle}); + if (listen) { + assert(socket->listen() == 0); + socket->set_handler(http->get_connection_handler()); + socket->start_loop(); + } + } + + ~BlobServer() { + socket->terminate(); + delete socket; + delete http; + } + + int handle(Request &req, Response &resp, std::string_view) { + ++requests; + last_target = std::string(req.target()); + assert(last_target.find("/blobs/" + digest) != std::string::npos); + size_t begin = 0, end = data.size() - 1; + auto range = std::string(req.headers["Range"]); + if (!range.empty()) { + assert(std::sscanf(range.c_str(), "bytes=%zu-%zu", &begin, &end) == 2); + end = std::min(end, data.size() - 1); + assert(begin <= end); + resp.set_result(206); + resp.headers.insert("Content-Range", "bytes " + std::to_string(begin) + "-" + + std::to_string(end) + "/" + std::to_string(data.size())); + } else { + resp.set_result(200); + } + resp.headers.content_length(end - begin + 1); + if (req.verb() != Verb::HEAD) + assert(resp.write(data.data() + begin, end - begin + 1) == (ssize_t)(end - begin + 1)); + return 0; + } +}; + +static void read_range(photon::fs::IFile *file, const BlobServer &blob, + size_t offset, size_t count) { + std::string bytes(count, '\0'); + assert(file->pread(&bytes[0], count, offset) == (ssize_t)count); + assert(bytes == blob.data.substr(offset, count)); +} + +static void test_routing(const std::string &root, const std::string &version, + bool cache_first, bool available, bool p2p_enabled = true) { + BlobServer origin, accelerator(available); + const auto blob_url = origin.url + "/v2/repo/blobs/" + origin.digest; + const auto config_path = root + "/config.json"; + const auto staging = root + "/blob"; + std::ofstream(staging, std::ios::binary) << origin.data; + // Exercise both configuration forms through actual initialization. + const auto cache_config = version == "v1" + ? "\"cacheType\":\"file\",\"registryCacheDir\":\"" + root + "/cache\",\"registryCacheSizeGB\":1," + : "\"cacheType\":\"ocf\",\"cacheConfig\":{\"cacheType\":\"file\",\"cacheDir\":\"" + root + + "/cache\",\"cacheSizeGB\":1},"; + std::ofstream(config_path) + << "{\"enableAudit\":false,\"logPath\":\"\",\"logLevel\":3," + << "\"registryFsVersion\":\"" << version << "\"," + << cache_config << "\"p2pConfig\":{\"enable\":" + << (p2p_enabled ? "true" : "false") << ",\"cacheFirst\":" + << (cache_first ? "true" : "false") << ",\"address\":\"" + << accelerator.url << "/accelerator\"}}"; + ImageService service(config_path.c_str()); + assert(service.init() == 0); + assert(service.global_fs.remote_fs == nullptr); + + auto import = [&] { + return service.import_local_blob(staging, blob_url, origin.digest, + origin.data.size()).status; + }; + const bool import_allowed = !p2p_enabled || cache_first; + // Import before the first image selects its read path. + assert(import() == (import_allowed ? CacheImportStatus::ok : CacheImportStatus::unsupported)); + assert(origin.requests == 0 && accelerator.requests == 0); + assert(service.enable_acceleration() == (p2p_enabled && available)); + if (cache_first || !p2p_enabled || !available) + assert(service.global_fs.remote_fs == service.global_fs.cached_fs); + else + assert(service.global_fs.remote_fs == service.global_fs.srcfs); + assert(import() == (import_allowed ? CacheImportStatus::ok : CacheImportStatus::unsupported)); + assert(::unlink(staging.c_str()) == 0); + + std::unique_ptr file( + service.global_fs.remote_fs->open(blob_url.c_str(), O_RDONLY)); + assert(file); + read_range(file.get(), origin, 0, origin.data.size()); + + if (import_allowed) { + // Imported bytes, including the unaligned tail, need no HTTP request. + assert(origin.requests == 0 && accelerator.requests == 0); + // Re-selection (as done by subsequent image creation) retains the cache. + assert(service.enable_acceleration() == (p2p_enabled && available)); + assert(service.global_fs.remote_fs == service.global_fs.cached_fs); + auto cached = dynamic_cast(file.get()); + assert(cached && cached->get_store()); + // Evict one aligned range; untouched ranges must still hit locally. + const size_t block = 256 * 1024; + assert(cached->get_store()->evict(block, block) == 0); + read_range(file.get(), origin, 0, block); + read_range(file.get(), origin, 2 * block, origin.data.size() - 2 * block); + assert(origin.requests == 0 && accelerator.requests == 0); + read_range(file.get(), origin, block, block); + if (p2p_enabled && available) { + assert(accelerator.requests > 0); + assert(accelerator.last_target.find("/accelerator/" + origin.url) == 0); + // RegistryFS v1 may obtain origin metadata before using P2P. + if (version == "v2") assert(origin.requests == 0); + } else { + assert(origin.requests > 0 && accelerator.requests == 0); + } + int origin_requests = origin.requests, p2p_requests = accelerator.requests; + read_range(file.get(), origin, 0, origin.data.size()); + assert(origin.requests == origin_requests && accelerator.requests == p2p_requests); + // A full eviction remains refillable, and refilled bytes hit thereafter. + assert(cached->get_store()->evict(0, origin.data.size()) == 0); + read_range(file.get(), origin, 0, origin.data.size()); + assert(origin.requests + accelerator.requests > origin_requests + p2p_requests); + origin_requests = origin.requests; + p2p_requests = accelerator.requests; + read_range(file.get(), origin, 0, origin.data.size()); + assert(origin.requests == origin_requests && accelerator.requests == p2p_requests); + } else if (available) { + // Default P2P behavior still bypasses the managed file cache. + int requests = accelerator.requests; + assert(requests > 0); + read_range(file.get(), origin, 0, origin.data.size()); + assert(accelerator.requests > requests); + } + std::cout << "PASS registry=" << version << " cacheFirst=" << cache_first + << " p2pEnabled=" << p2p_enabled << " available=" << available << std::endl; +} + +static int remove_entry(const char *path, const struct stat *, int, struct FTW *) { + return ::remove(path); +} + +int main() { + assert(photon::init(photon::INIT_EVENT_EPOLL, photon::INIT_IO_DEFAULT) == 0); + struct Mode { bool cache_first, available, enabled; }; + const Mode modes[] = { + {true, true, true}, // Cache first with an available accelerator. + {false, true, true}, // Existing cache bypass. + {true, false, true}, // Cache first falls back to the origin. + {false, false, true}, // Bypass policy still rejects early imports. + {true, true, false}, // P2P disabled. + }; + for (const auto &version : {"v2", "v1"}) { + for (const auto &mode : modes) { + char root[] = "/tmp/p2p-cache-test-XXXXXX"; + assert(mkdtemp(root)); + test_routing(root, version, mode.cache_first, mode.available, mode.enabled); + assert(::nftw(root, remove_entry, 32, FTW_DEPTH | FTW_PHYS) == 0); + } + } + for (const auto &type : {"ocf", "download"}) { + char root[] = "/tmp/p2p-cache-config-XXXXXX"; + assert(mkdtemp(root)); + auto path = std::string(root) + "/config.json"; + std::ofstream(path) << "{\"enableAudit\":false,\"logPath\":\"\",\"cacheType\":\"" + << type << "\",\"p2pConfig\":{\"cacheFirst\":true}}"; + { ImageService service(path.c_str()); assert(service.init() != 0); } + assert(::nftw(root, remove_entry, 32, FTW_DEPTH | FTW_PHYS) == 0); + } + photon::fini(); +} diff --git a/src/ublk/test/cache_import_test.cpp b/src/ublk/test/cache_import_test.cpp index 850ba427..4c8594ce 100644 --- a/src/ublk/test/cache_import_test.cpp +++ b/src/ublk/test/cache_import_test.cpp @@ -155,6 +155,22 @@ int main() { auto first = make_blob(source, small, 42); assert(import(service, source + "/" + first, first, small) == 200); assert(src->opens == 0); + // Admission must resolve the same cache type as ImageService::init. + for (const auto &config : {"{}", "{\"cacheType\":\"file\"}", + "{\"cacheType\":\"file\",\"cacheConfig\":{\"cacheType\":\"\"}}", + "{\"cacheType\":\"ocf\",\"cacheConfig\":{\"cacheType\":\"file\"}}"}) { + std::ofstream(configPath) << config; + assert(service.global_conf.ParseJSON(configPath)); + assert(import(service, source + "/" + first, first, small) == 200); + } + for (const auto &config : {"{\"cacheType\":\"ocf\"}", + "{\"cacheType\":\"file\",\"cacheConfig\":{\"cacheType\":\"download\"}}"}) { + std::ofstream(configPath) << config; + assert(service.global_conf.ParseJSON(configPath)); + assert(import(service, source + "/" + first, first, small) == 409); + } + std::ofstream(configPath) << "{\"cacheConfig\":{\"cacheType\":\"file\"}}"; + assert(service.global_conf.ParseJSON(configPath)); // Reject invalid requests before touching the managed cache. int status = 0; std::string message; From 9e01600b55400af74b53c51bc0d17ca5eaac4dc0 Mon Sep 17 00:00:00 2001 From: Wenxiang Jin Date: Mon, 9 Nov 2026 11:04:04 +0800 Subject: [PATCH 4/4] Rerun CI for cache-first P2P imports Signed-off-by: Wenxiang Jin