feat: import local blobs with optional cache-first P2P reads

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:<P2PConfig.Port>/<P2PConfig.APIKey>`, depending on dadip2p.yaml |
 | exporterConfig.enable         | whether or not create a server to show Prometheus metrics.                                  |
 | exporterConfig.uriPrefix      | URI prefix for export metrics.                                                              |

docs/standalone-usage.md

@@ -111,6 +111,46 @@ 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, 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/<digest>` 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. 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

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

src/cache_import.cpp

@@ -0,0 +1,135 @@
+/*
+   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 <algorithm>
+#include <array>
+#include <cerrno>
+#include <fcntl.h>
+#include <sys/stat.h>
+#include <unistd.h>
+#include <vector>
+#include <openssl/sha.h>
+#include <photon/thread/thread.h>
+
+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.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);
+    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<unsigned char> buf(256 * 1024);
+    std::vector<std::array<unsigned char, SHA256_DIGEST_LENGTH>> 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<int64_t>(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<unsigned char, SHA256_DIGEST_LENGTH> 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<photon::fs::IFile> 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<int64_t>(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<unsigned char, SHA256_DIGEST_LENGTH> 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, ""};
+}

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 {

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();

src/image_service.h

@@ -17,6 +17,8 @@
 #pragma once
 
 #include <string>
+#include <atomic>
+#include <cstdint>
 #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<std::string, std::string> reload_auth(const char *remote_path);
     void set_result_file(std::string &filename, std::string &data);

src/test/CMakeLists.txt

@@ -74,3 +74,34 @@ 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)
+
+# 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)