Restier tiered storage (#260) * add restier interface and ipfs implementation * integrate ipfs, config changes and test script * Integrated migration into tiered storage * implemented secondary index logic * add feature to remove from ipfs if value is updated for a key * Fix storage test build and tiered storage tests
diff --git a/chain/storage/BUILD b/chain/storage/BUILD index 45aa6e9..313c738 100644 --- a/chain/storage/BUILD +++ b/chain/storage/BUILD
@@ -55,6 +55,53 @@ "//common/lru:lru_cache", "//platform/statistic:stats", "//third_party:leveldb", + "@com_google_absl//absl/base", + "@com_google_absl//absl/synchronization", + ], +) + +cc_library( + name = "ipfs_client", + srcs = ["ipfs_client.cpp"], + hdrs = ["ipfs_client.h"], + deps = [ + ":storage", + "//chain/storage/proto:ipfs_config_cc_proto", + "//common:comm", + "@boost//:asio", + "@boost//:beast", + ], +) + +cc_library( + name = "secondary_index", + hdrs = ["interface/secondary_index.h"], + deps = [], +) + +cc_library( + name = "in_memory_index", + srcs = ["in_memory_index.cpp"], + hdrs = ["in_memory_index.h"], + deps = [ + ":secondary_index", + ], +) + +cc_library( + name = "tiered_storage", + srcs = ["tiered_storage.cpp"], + hdrs = ["tiered_storage.h"], + deps = [ + ":in_memory_index", + ":leveldb", + ":storage", + ":ipfs_client", + "//chain/storage/proto:ipfs_config_cc_proto", + "//chain/storage/proto:tiered_index_cc_proto", + "//common:comm", + "@com_google_absl//absl/base", + "@com_google_absl//absl/synchronization", ], ) @@ -82,6 +129,20 @@ ], ) +cc_test( + name = "tiered_storage_test", + size = "small", + timeout = "short", + srcs = ["tiered_storage_test.cpp"], + deps = [ + ":leveldb", + ":memory_db", + ":tiered_storage", + ":ipfs_client", + "//common/test:test_main", + ], +) + cc_library( name = "duckdb_storage", srcs = ["duckdb.cpp"], @@ -95,6 +156,15 @@ ) cc_binary( + name = "ipfs_integration_test", + srcs = ["ipfs_integration_test.cpp"], + deps = [ + ":ipfs_client", + "//chain/storage/proto:ipfs_config_cc_proto", + ], +) + +cc_binary( name = "duckdb_test", srcs = ["duckdb_test.cpp"], deps = [ @@ -103,4 +173,4 @@ "//common:glog", "//chain/storage/proto:duckdb_config_cc_proto", ], -) \ No newline at end of file +)
diff --git a/chain/storage/in_memory_index.cpp b/chain/storage/in_memory_index.cpp new file mode 100644 index 0000000..f25098e --- /dev/null +++ b/chain/storage/in_memory_index.cpp
@@ -0,0 +1,46 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one + * or more contributor license agreements. See the NOTICE file + * distributed with this work for additional information + * regarding copyright ownership. The ASF licenses this file + * to you 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 "chain/storage/in_memory_index.h" + +namespace resdb { +namespace storage { + +void InMemoryHashIndex::Add(const std::string& key, const std::string& cid) { + index_[key] = cid; +} + +std::string InMemoryHashIndex::Get(const std::string& key) const { + auto it = index_.find(key); + if (it != index_.end()) { + return it->second; + } + return ""; +} + +void InMemoryHashIndex::Clear() { + index_.clear(); +} + +size_t InMemoryHashIndex::Size() const { + return index_.size(); +} + +} // namespace storage +} // namespace resdb
diff --git a/chain/storage/in_memory_index.h b/chain/storage/in_memory_index.h new file mode 100644 index 0000000..307e482 --- /dev/null +++ b/chain/storage/in_memory_index.h
@@ -0,0 +1,42 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one + * or more contributor license agreements. See the NOTICE file + * distributed with this work for additional information + * regarding copyright ownership. The ASF licenses this file + * to you 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 <string> +#include <unordered_map> + +#include "chain/storage/interface/secondary_index.h" + +namespace resdb { +namespace storage { + +class InMemoryHashIndex : public SecondaryIndex { + public: + void Add(const std::string& key, const std::string& cid) override; + std::string Get(const std::string& key) const override; + void Clear() override; + size_t Size() const override; + + private: + std::unordered_map<std::string, std::string> index_; +}; + +} // namespace storage +} // namespace resdb
diff --git a/chain/storage/interface/secondary_index.h b/chain/storage/interface/secondary_index.h new file mode 100644 index 0000000..91d6cc6 --- /dev/null +++ b/chain/storage/interface/secondary_index.h
@@ -0,0 +1,37 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one + * or more contributor license agreements. See the NOTICE file + * distributed with this work for additional information + * regarding copyright ownership. The ASF licenses this file + * to you 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 <string> + +namespace resdb { +namespace storage { + +class SecondaryIndex { + public: + virtual ~SecondaryIndex() = default; + virtual void Add(const std::string& key, const std::string& cid) = 0; + virtual std::string Get(const std::string& key) const = 0; + virtual void Clear() = 0; + virtual size_t Size() const = 0; +}; + +} // namespace storage +} // namespace resdb
diff --git a/chain/storage/ipfs_client.cpp b/chain/storage/ipfs_client.cpp new file mode 100644 index 0000000..7ec5921 --- /dev/null +++ b/chain/storage/ipfs_client.cpp
@@ -0,0 +1,392 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one + * or more contributor license agreements. See the NOTICE file + * distributed with this work for additional information + * regarding copyright ownership. The ASF licenses this file + * to you 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 "chain/storage/ipfs_client.h" + +#include <boost/asio.hpp> +#include <boost/beast.hpp> + +#include <glog/logging.h> +#include <regex> +#include <sstream> +#include <thread> + +namespace resdb { +namespace storage { + +namespace asio = boost::asio; +namespace beast = boost::beast; +using tcp = asio::ip::tcp; + +namespace { + +constexpr int64_t kDefaultTimeoutMs = 30000; +constexpr int kDefaultMaxRetries = 3; +constexpr char kBoundary[] = "ipfs-client-boundary-1a2b3c4d"; + +struct EndpointInfo { + std::string host; + uint16_t port; +}; + +EndpointInfo ParseEndpoint(const std::string& url) { + EndpointInfo info{"localhost", 5001}; + std::string cleaned = url; + if (cleaned.find("http://") == 0) { + cleaned = cleaned.substr(7); + } else if (cleaned.find("https://") == 0) { + cleaned = cleaned.substr(8); + } + auto colon_pos = cleaned.find(':'); + if (colon_pos != std::string::npos) { + info.host = cleaned.substr(0, colon_pos); + try { + info.port = static_cast<uint16_t>(std::stoi(cleaned.substr(colon_pos + 1))); + } catch (...) { + info.port = 5001; + } + } else { + info.host = cleaned; + } + return info; +} + +std::string ExtractHashFromJson(const std::string& json) { + std::regex hash_regex("\"Hash\"\\s*:\\s*\"([^\"]+)\""); + std::smatch match; + if (std::regex_search(json, match, hash_regex) && match.size() > 1) { + return match[1].str(); + } + std::regex cid_regex("\"Cid\"\\s*:\\s*\\{\\s*\"/\"\\s*:\\s*\"([^\"]+)\""); + if (std::regex_search(json, match, cid_regex) && match.size() > 1) { + return match[1].str(); + } + return ""; +} + +std::string BuildMultipartBody(const std::string& data, const std::string& filename = "data") { + std::ostringstream oss; + oss << "--" << kBoundary << "\r\n" + << "Content-Disposition: form-data; name=\"file\"; filename=\"" << filename << "\"\r\n" + << "Content-Type: application/octet-stream\r\n" + << "\r\n" + << data << "\r\n" + << "--" << kBoundary << "--\r\n"; + return oss.str(); +} + +class IPFSHttpClient { + public: + IPFSHttpClient(const EndpointInfo& endpoint, int64_t timeout_ms, int max_retries) + : endpoint_(endpoint), timeout_ms_(timeout_ms), max_retries_(max_retries) { + LOG(INFO) << "IPFS HTTP client initialized: host=" << endpoint.host + << " port=" << endpoint.port + << " timeout=" << timeout_ms << "ms" + << " max_retries=" << max_retries; + } + + std::string PostRaw(const std::string& path, const std::string& body, + const std::string& content_type = "application/octet-stream") { + beast::error_code ec; + beast::flat_buffer buffer; + beast::http::response<beast::http::string_body> response; + + for (int attempt = 0; attempt <= max_retries_; ++attempt) { + try { + tcp::resolver resolver(io_context_); + tcp::socket socket(io_context_); + + auto const results = resolver.resolve(endpoint_.host, + std::to_string(endpoint_.port), ec); + if (ec) { + LOG(ERROR) << "IPFS resolve failed (attempt " << attempt + 1 << "/" + << max_retries_ + 1 << "): " << ec.message(); + if (attempt < max_retries_) { + std::this_thread::sleep_for(std::chrono::milliseconds(500 * (attempt + 1))); + continue; + } + return ""; + } + + beast::get_lowest_layer(socket).connect(*results.begin(), ec); + if (ec) { + LOG(ERROR) << "IPFS connect failed (attempt " << attempt + 1 << "/" + << max_retries_ + 1 << "): " << ec.message(); + if (attempt < max_retries_) { + std::this_thread::sleep_for(std::chrono::milliseconds(500 * (attempt + 1))); + continue; + } + return ""; + } + + beast::http::request<beast::http::string_body> req{ + beast::http::verb::post, path, 11}; + req.set(beast::http::field::host, + endpoint_.host + ":" + std::to_string(endpoint_.port)); + req.set(beast::http::field::content_type, content_type); + req.set(beast::http::field::user_agent, "resdb-ipfs-client/1.0"); + req.body() = body; + req.prepare_payload(); + + beast::http::write(socket, req, ec); + if (ec) { + LOG(ERROR) << "IPFS write failed (attempt " << attempt + 1 << "/" + << max_retries_ + 1 << "): " << ec.message(); + if (attempt < max_retries_) { + std::this_thread::sleep_for(std::chrono::milliseconds(500 * (attempt + 1))); + continue; + } + return ""; + } + + beast::http::read(socket, buffer, response, ec); + if (ec) { + LOG(ERROR) << "IPFS read failed (attempt " << attempt + 1 << "/" + << max_retries_ + 1 << "): " << ec.message(); + if (attempt < max_retries_) { + std::this_thread::sleep_for(std::chrono::milliseconds(500 * (attempt + 1))); + continue; + } + return ""; + } + + socket.shutdown(tcp::socket::shutdown_both, ec); + + if (response.result() != beast::http::status::ok) { + LOG(ERROR) << "IPFS HTTP error: " << response.result() + << " body: " << response.body(); + return ""; + } + + return response.body(); + } catch (const std::exception& e) { + LOG(ERROR) << "IPFS request exception (attempt " << attempt + 1 << "/" + << max_retries_ + 1 << "): " << e.what(); + if (attempt < max_retries_) { + std::this_thread::sleep_for(std::chrono::milliseconds(500 * (attempt + 1))); + continue; + } + return ""; + } + } + + return ""; + } + + private: + EndpointInfo endpoint_; + int64_t timeout_ms_; + int max_retries_; + asio::io_context io_context_; +}; + +std::unique_ptr<IPFSHttpClient> CreateHttpClient(const IPFSConfig& config, + const std::string& endpoint_override = "") { + std::string endpoint = endpoint_override.empty() ? config.api_endpoint() : endpoint_override; + if (endpoint.empty()) { + endpoint = "http://localhost:5001"; + } + auto info = ParseEndpoint(endpoint); + int64_t timeout = config.timeout_ms() > 0 ? config.timeout_ms() : kDefaultTimeoutMs; + int retries = config.max_retries() > 0 ? config.max_retries() : kDefaultMaxRetries; + return std::make_unique<IPFSHttpClient>(info, timeout, retries); +} + +} // namespace + +class IPFSClientImpl : public IPFSClient { + public: + explicit IPFSClientImpl(const IPFSConfig& config) + : config_(config), + enabled_(config.enabled()) { + if (!enabled_) { + LOG(WARNING) << "IPFS client created but not enabled"; + } + if (config_.api_endpoint().empty()) { + config_.set_api_endpoint("http://localhost:5001"); + } + if (config_.gateway_endpoint().empty()) { + config_.set_gateway_endpoint("http://localhost:8080"); + } + LOG(INFO) << "IPFS client initialized with endpoint: " << config_.api_endpoint(); + } + + std::string Add(const std::string& data) override { + if (!enabled_) { + LOG(ERROR) << "IPFS is not enabled, cannot add data"; + return ""; + } + + LOG(INFO) << "IPFS Add: data size = " << data.size(); + + std::string body = BuildMultipartBody(data); + std::string content_type = std::string("multipart/form-data; boundary=") + kBoundary; + + std::string response = PostWithRetry("/api/v0/add", body, content_type); + if (response.empty()) { + LOG(ERROR) << "IPFS Add: failed to add data"; + return ""; + } + + std::string cid = ExtractHashFromJson(response); + if (cid.empty()) { + LOG(ERROR) << "IPFS Add: could not extract CID from response: " << response; + return ""; + } + + LOG(INFO) << "IPFS Add: successfully added data, CID = " << cid; + return cid; + } + + std::string AddDAG(const std::string& json_data) override { + if (!enabled_) { + LOG(ERROR) << "IPFS is not enabled, cannot add DAG"; + return ""; + } + + LOG(INFO) << "IPFS AddDAG: json size = " << json_data.size(); + + std::string body = BuildMultipartBody(json_data, "dag.json"); + std::string content_type = std::string("multipart/form-data; boundary=") + kBoundary; + + std::string response = PostWithRetry("/api/v0/dag/put", body, content_type); + if (response.empty()) { + LOG(ERROR) << "IPFS AddDAG: failed to add DAG node"; + return ""; + } + + std::string cid = ExtractHashFromJson(response); + if (cid.empty()) { + LOG(ERROR) << "IPFS AddDAG: could not extract CID from response: " << response; + return ""; + } + + LOG(INFO) << "IPFS AddDAG: successfully added DAG node, CID = " << cid; + return cid; + } + + std::string Cat(const std::string& cid) override { + if (!enabled_) { + LOG(ERROR) << "IPFS is not enabled, cannot cat data"; + return ""; + } + + LOG(INFO) << "IPFS Cat: CID = " << cid; + + std::string path = std::string("/api/v0/cat?arg=") + cid; + std::string response = PostWithRetry(path, "", "application/octet-stream"); + if (response.empty()) { + LOG(ERROR) << "IPFS Cat: failed to retrieve data for CID = " << cid; + return ""; + } + + LOG(INFO) << "IPFS Cat: successfully retrieved data, size = " << response.size(); + return response; + } + + std::string GetDAG(const std::string& cid) override { + if (!enabled_) { + LOG(ERROR) << "IPFS is not enabled, cannot get DAG"; + return ""; + } + + LOG(INFO) << "IPFS GetDAG: CID = " << cid; + + std::string path = std::string("/api/v0/dag/get?arg=") + cid; + std::string response = PostWithRetry(path, "", "application/octet-stream"); + if (response.empty()) { + LOG(ERROR) << "IPFS GetDAG: failed to retrieve DAG node for CID = " << cid; + return ""; + } + + LOG(INFO) << "IPFS GetDAG: successfully retrieved DAG node"; + return response; + } + + bool Exists(const std::string& cid) override { + if (!enabled_) { + return false; + } + + LOG(INFO) << "IPFS Exists: CID = " << cid; + + std::string path = std::string("/api/v0/pin/ls?arg=") + cid + "&type=recursive"; + std::string response = PostWithRetry(path, "", "application/octet-stream"); + bool exists = !response.empty() && response.find("\"Type\":\"recursive\"") != std::string::npos; + + if (exists) { + LOG(INFO) << "IPFS Exists: CID " << cid << " exists"; + } else { + LOG(INFO) << "IPFS Exists: CID " << cid << " does not exist"; + } + return exists; + } + + bool Unpin(const std::string& cid) override { + if (!enabled_) { + LOG(ERROR) << "IPFS is not enabled, cannot unpin"; + return false; + } + + LOG(INFO) << "IPFS Unpin: CID = " << cid; + + std::string path = std::string("/api/v0/pin/rm?arg=") + cid; + std::string response = PostWithRetry(path, "", "application/octet-stream"); + bool success = !response.empty() && response.find("\"Pins\"") != std::string::npos; + + if (success) { + LOG(INFO) << "IPFS Unpin: successfully unpinned CID = " << cid; + } else { + LOG(ERROR) << "IPFS Unpin: failed to unpin CID = " << cid + << " response: " << response; + } + return success; + } + + bool IsEnabled() const override { return enabled_; } + + private: + std::string PostWithRetry(const std::string& path, const std::string& body, + const std::string& content_type) { + auto client = CreateHttpClient(config_); + return client->PostRaw(path, body, content_type); + } + + IPFSConfig config_; + bool enabled_; +}; + +std::unique_ptr<IPFSClient> IPFSClient::Create(const IPFSConfig& config) { + return std::make_unique<IPFSClientImpl>(config); +} + +std::unique_ptr<IPFSClient> NewIPFSClient( + const std::string& api_endpoint, + bool enabled) { + IPFSConfig config; + config.set_api_endpoint(api_endpoint); + config.set_enabled(enabled); + config.set_gateway_endpoint("http://localhost:8080"); + config.set_timeout_ms(kDefaultTimeoutMs); + config.set_max_retries(kDefaultMaxRetries); + return std::make_unique<IPFSClientImpl>(config); +} + +} // namespace storage +} // namespace resdb
diff --git a/chain/storage/ipfs_client.h b/chain/storage/ipfs_client.h new file mode 100644 index 0000000..8eca553 --- /dev/null +++ b/chain/storage/ipfs_client.h
@@ -0,0 +1,53 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one + * or more contributor license agreements. See the NOTICE file + * distributed with this work for additional information + * regarding copyright ownership. The ASF licenses this file + * to you 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 <memory> +#include <optional> +#include <string> +#include <vector> + +#include "chain/storage/proto/ipfs_config.pb.h" + +namespace resdb { +namespace storage { + +class IPFSClient { + public: + virtual ~IPFSClient() = default; + + virtual std::string Add(const std::string& data) = 0; + virtual std::string AddDAG(const std::string& json_data) = 0; + virtual std::string Cat(const std::string& cid) = 0; + virtual std::string GetDAG(const std::string& cid) = 0; + virtual bool Exists(const std::string& cid) = 0; + virtual bool Unpin(const std::string& cid) = 0; + virtual bool IsEnabled() const = 0; + + static std::unique_ptr<IPFSClient> Create( + const IPFSConfig& config); +}; + +std::unique_ptr<IPFSClient> NewIPFSClient( + const std::string& api_endpoint, + bool enabled = true); + +} // namespace storage +} // namespace resdb \ No newline at end of file
diff --git a/chain/storage/ipfs_integration_test.cpp b/chain/storage/ipfs_integration_test.cpp new file mode 100644 index 0000000..b5e3d80 --- /dev/null +++ b/chain/storage/ipfs_integration_test.cpp
@@ -0,0 +1,39 @@ +#include <glog/logging.h> +#include <iostream> +#include "chain/storage/ipfs_client.h" +#include "chain/storage/proto/ipfs_config.pb.h" + +using namespace resdb::storage; + +int main() { + google::InitGoogleLogging("ipfs_test"); + FLAGS_minloglevel = 0; + + IPFSConfig config; + config.set_api_endpoint("127.0.0.1:5001"); + config.set_enabled(true); + config.set_gateway_endpoint("127.0.0.1:8080"); + config.set_timeout_ms(30000); + config.set_max_retries(3); + + auto client = IPFSClient::Create(config); + std::cout << "IPFS client created, enabled=" << client->IsEnabled() << std::endl; + + std::string cid = client->Add("hello_tiered_storage"); + std::cout << "Add returned CID: " << cid << std::endl; + + std::string data = client->Cat(cid); + std::cout << "Cat returned: " << data << std::endl; + + bool exists = client->Exists(cid); + std::cout << "Exists(" << cid << ")=" << (exists ? "true" : "false") << std::endl; + + std::string dag_cid = client->AddDAG("{\"key\":\"test_key_1\",\"value\":\"hello_tiered_storage\"}"); + std::cout << "AddDAG returned CID: " << dag_cid << std::endl; + + std::string dag_data = client->GetDAG(dag_cid); + std::cout << "GetDAG returned: " << dag_data << std::endl; + + std::cout << "IPFS test complete!" << std::endl; + return 0; +}
diff --git a/chain/storage/leveldb.cpp b/chain/storage/leveldb.cpp index de49376..cfdd57c 100644 --- a/chain/storage/leveldb.cpp +++ b/chain/storage/leveldb.cpp
@@ -19,6 +19,7 @@ #include "chain/storage/leveldb.h" +#include <chrono> #include <glog/logging.h> #include <unistd.h> @@ -159,9 +160,12 @@ } int ResLevelDB::SetValue(const std::string& key, const std::string& value) { + auto start = std::chrono::high_resolution_clock::now(); if (block_cache_) { block_cache_->Put(key, value); } + + absl::MutexLock lock(&batch_mutex_); batch_.Put(key, value); if (batch_.ApproximateSize() >= write_batch_size_) { @@ -169,16 +173,23 @@ if (status.ok()) { batch_.Clear(); UpdateMetrics(); + auto end = std::chrono::high_resolution_clock::now(); + auto us = std::chrono::duration_cast<std::chrono::microseconds>(end - start).count(); + LOG(INFO) << "[timer] ldb_set key=" << key << " us=" << us; return 0; } else { LOG(ERROR) << "flush buffer fail:" << status.ToString(); return -1; } } + auto end = std::chrono::high_resolution_clock::now(); + auto us = std::chrono::duration_cast<std::chrono::microseconds>(end - start).count(); + LOG(INFO) << "[timer] ldb_set key=" << key << " us=" << us; return 0; } std::string ResLevelDB::GetValue(const std::string& key) { + auto start = std::chrono::high_resolution_clock::now(); std::string value; bool found_in_cache = false; @@ -190,11 +201,14 @@ if (!found_in_cache) { leveldb::Status status = db_->Get(leveldb::ReadOptions(), key, &value); if (!status.ok()) { - value.clear(); // Ensure value is empty if not found in DB + value.clear(); } } UpdateMetrics(); + auto end = std::chrono::high_resolution_clock::now(); + auto us = std::chrono::duration_cast<std::chrono::microseconds>(end - start).count(); + LOG(INFO) << "[timer] ldb_get key=" << key << " us=" << us; return value; } @@ -229,6 +243,7 @@ } bool ResLevelDB::Flush() { + absl::MutexLock lock(&batch_mutex_); leveldb::Status status = db_->Write(leveldb::WriteOptions(), &batch_); if (status.ok()) { batch_.Clear(); @@ -425,6 +440,23 @@ return SetValue(ckpt_key, std::to_string(ckpt)); } +bool ResLevelDB::DeleteKey(const std::string& key) { + absl::MutexLock lock(&batch_mutex_); + batch_.Delete(key); + leveldb::Status status = db_->Write(leveldb::WriteOptions(), &batch_); + if (status.ok()) { + batch_.Clear(); + if (block_cache_) { + block_cache_->Remove(key); + } + LOG(INFO) << "Deleted key from LevelDB: " << key; + return true; + } + LOG(ERROR) << "Failed to delete key from LevelDB: " << key + << ", error: " << status.ToString(); + return false; +} + uint64_t ResLevelDB::GetLastCheckpointInternal() { std::string value = GetValue(ckpt_key); if (value.empty()) {
diff --git a/chain/storage/leveldb.h b/chain/storage/leveldb.h index 67ec7a4..3335611 100644 --- a/chain/storage/leveldb.h +++ b/chain/storage/leveldb.h
@@ -29,6 +29,8 @@ #include "leveldb/db.h" #include "leveldb/write_batch.h" #include "platform/statistic/stats.h" +#include "absl/base/thread_annotations.h" +#include "absl/synchronization/mutex.h" namespace resdb { namespace storage { @@ -38,7 +40,7 @@ std::unique_ptr<Storage> NewResLevelDB( std::optional<LevelDBInfo> config = std::nullopt); -class ResLevelDB : public Storage { +class ResLevelDB : public Storage, public DeletableStorage { public: ResLevelDB(std::optional<LevelDBInfo> config_data = std::nullopt); @@ -80,6 +82,8 @@ virtual int SetLastCheckpoint(uint64_t ckpt); + bool DeleteKey(const std::string& key) override; + private: void CreateDB(const std::string& path); uint64_t GetLastCheckpointInternal(); @@ -88,6 +92,7 @@ private: std::unique_ptr<leveldb::DB> db_ = nullptr; ::leveldb::WriteBatch batch_; + absl::Mutex batch_mutex_; unsigned int write_buffer_size_ = 64 << 20; unsigned int write_batch_size_ = 1;
diff --git a/chain/storage/memory_db.cpp b/chain/storage/memory_db.cpp index aac3bc8..158be3b 100644 --- a/chain/storage/memory_db.cpp +++ b/chain/storage/memory_db.cpp
@@ -19,6 +19,7 @@ #include "chain/storage/memory_db.h" +#include <chrono> #include <glog/logging.h> namespace resdb { @@ -50,46 +51,56 @@ std::string MemoryDB::GetValue(const std::string& key) { auto search = kv_map_.find(key); - if (search != kv_map_.end()) - return search->second; - else { - return ""; - } + return (search != kv_map_.end()) ? search->second : ""; } std::pair<std::string, uint64_t> MemoryDB::GetValueWithSeq( const std::string& key, uint64_t seq) { - auto search_it = kv_map_with_seq_.find(key); - if (search_it != kv_map_with_seq_.end() && search_it->second.size()) { - auto it = search_it->second.end(); - do { - --it; - if (it->second == seq || seq == 0) { - return *it; - } - if (it->second < seq) { - break; - } - } while (it != search_it->second.begin()); - LOG(ERROR) << " key:" << key << " no seq:" << seq; - } - return std::make_pair("", 0); + auto start = std::chrono::high_resolution_clock::now(); + auto result = [&]() -> std::pair<std::string, uint64_t> { + auto search_it = kv_map_with_seq_.find(key); + if (search_it != kv_map_with_seq_.end() && search_it->second.size()) { + auto it = search_it->second.end(); + do { + --it; + if (it->second == seq || seq == 0) { + return *it; + } + if (it->second < seq) { + break; + } + } while (it != search_it->second.begin()); + LOG(ERROR) << " key:" << key << " no seq:" << seq; + } + return std::make_pair("", 0); + }(); + auto end = std::chrono::high_resolution_clock::now(); + auto us = std::chrono::duration_cast<std::chrono::microseconds>(end - start).count(); + LOG(INFO) << "[timer] memdb_get_seq key=" << key << " us=" << us; + return result; } int MemoryDB::SetValueWithSeq(const std::string& key, const std::string& value, uint64_t seq) { - auto it = kv_map_with_seq_.find(key); - if (it != kv_map_with_seq_.end() && it->second.back().second > seq) { - LOG(ERROR) << " value seq not match. key:" << key << " db seq:" - << (it == kv_map_with_seq_.end() ? 0 : it->second.back().second) - << " new seq:" << seq; - return -2; - } - kv_map_with_seq_[key].push_back(std::make_pair(value, seq)); - while (kv_map_with_seq_[key].size() > max_history_) { - kv_map_with_seq_[key].erase(kv_map_with_seq_[key].begin()); - } - return 0; + auto start = std::chrono::high_resolution_clock::now(); + int ret = [&]() { + auto it = kv_map_with_seq_.find(key); + if (it != kv_map_with_seq_.end() && it->second.back().second > seq) { + LOG(ERROR) << " value seq not match. key:" << key << " db seq:" + << (it == kv_map_with_seq_.end() ? 0 : it->second.back().second) + << " new seq:" << seq; + return -2; + } + kv_map_with_seq_[key].push_back(std::make_pair(value, seq)); + while (kv_map_with_seq_[key].size() > max_history_) { + kv_map_with_seq_[key].erase(kv_map_with_seq_[key].begin()); + } + return 0; + }(); + auto end = std::chrono::high_resolution_clock::now(); + auto us = std::chrono::duration_cast<std::chrono::microseconds>(end - start).count(); + LOG(INFO) << "[timer] memdb_set_seq key=" << key << " us=" << us; + return ret; } int MemoryDB::SetValueWithVersion(const std::string& key, @@ -201,5 +212,12 @@ return resp; } +bool MemoryDB::DeleteKey(const std::string& key) { + kv_map_.erase(key); + kv_map_with_seq_.erase(key); + kv_map_with_v_.erase(key); + return true; +} + } // namespace storage } // namespace resdb
diff --git a/chain/storage/memory_db.h b/chain/storage/memory_db.h index ec138c7..e944f72 100644 --- a/chain/storage/memory_db.h +++ b/chain/storage/memory_db.h
@@ -49,7 +49,7 @@ std::unique_ptr<Storage> NewMemoryDB(); -class MemoryDB : public Storage { +class MemoryDB : public Storage, public DeletableStorage { public: MemoryDB(); @@ -83,6 +83,8 @@ std::vector<std::pair<std::string, int>> GetTopHistory(const std::string& key, int number) override; + bool DeleteKey(const std::string& key) override; + private: std::unordered_map<std::string, std::string> kv_map_; std::unordered_map<std::string, std::list<std::pair<std::string, int>>>
diff --git a/chain/storage/proto/BUILD b/chain/storage/proto/BUILD index 2558977..f81a61b 100644 --- a/chain/storage/proto/BUILD +++ b/chain/storage/proto/BUILD
@@ -44,6 +44,26 @@ ) proto_library( + name = "ipfs_config_proto", + srcs = ["ipfs_config.proto"], +) + +cc_proto_library( + name = "ipfs_config_cc_proto", + deps = [":ipfs_config_proto"], +) + +proto_library( + name = "tiered_index_proto", + srcs = ["tiered_index.proto"], +) + +cc_proto_library( + name = "tiered_index_cc_proto", + deps = [":tiered_index_proto"], +) + +proto_library( name = "duckdb_config_proto", srcs = ["duckdb_config.proto"], ) @@ -52,4 +72,3 @@ name = "duckdb_config_cc_proto", deps = [":duckdb_config_proto"], ) -
diff --git a/chain/storage/proto/ipfs_config.proto b/chain/storage/proto/ipfs_config.proto new file mode 100644 index 0000000..3a95918 --- /dev/null +++ b/chain/storage/proto/ipfs_config.proto
@@ -0,0 +1,46 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one + * or more contributor license agreements. See the NOTICE file + * distributed with this work for additional information + * regarding copyright ownership. The ASF licenses this file + * to you 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. + */ + +syntax = "proto3"; + +package resdb.storage; + +message IPFSConfig { + string api_endpoint = 1; + bool enabled = 2; + string gateway_endpoint = 3; + int32 timeout_ms = 4; + int32 max_retries = 5; +} + +message TieredStorageConfig { + enum HotStorageBackend { + MEMORYDB = 0; + LEVELDB = 1; + } + + int32 cold_threshold_checkpoint = 1; + bool enabled = 2; + int32 poll_interval_seconds = 3; + int32 batch_size = 4; + int64 max_cold_storage_bytes = 5; + bool auto_migration_enabled = 6; + HotStorageBackend hot_backend = 7; + int32 checkpoint_watermark = 8; +} \ No newline at end of file
diff --git a/chain/storage/proto/tiered_index.proto b/chain/storage/proto/tiered_index.proto new file mode 100644 index 0000000..79a314e --- /dev/null +++ b/chain/storage/proto/tiered_index.proto
@@ -0,0 +1,55 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one + * or more contributor license agreements. See the NOTICE file + * distributed with this work for additional information + * regarding copyright ownership. The ASF licenses this file + * to you 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. + */ + +syntax = "proto3"; + +package resdb.storage; + +message TieredIndexEntry { + string key = 1; + string key_range_start = 2; + string key_range_end = 3; + string ipfs_cid = 4; + uint64 checkpoint_height = 5; + int64 migrated_at_timestamp = 6; + bool has_versions = 7; + string versions_cid = 8; +} + +message TieredIndex { + repeated TieredIndexEntry entries = 1; + int32 version = 2; + int64 last_updated_timestamp = 3; +} + +message IndexManifest { + message RangeMapping { + string start_key = 1; + string end_key = 2; + string ipfs_cid = 3; + uint64 min_checkpoint = 4; + uint64 max_checkpoint = 5; + } + + repeated RangeMapping range_mappings = 1; + uint64 total_keys = 2; + uint64 cold_keys = 3; + int32 version = 4; + int64 last_updated_timestamp = 5; +} \ No newline at end of file
diff --git a/chain/storage/storage.h b/chain/storage/storage.h index 7a0bfe5..94b4101 100644 --- a/chain/storage/storage.h +++ b/chain/storage/storage.h
@@ -25,6 +25,12 @@ namespace resdb { +class DeletableStorage { + public: + virtual ~DeletableStorage() = default; + virtual bool DeleteKey(const std::string& key) = 0; +}; + class Storage { public: Storage() = default;
diff --git a/chain/storage/tiered_storage.cpp b/chain/storage/tiered_storage.cpp new file mode 100644 index 0000000..be2fb21 --- /dev/null +++ b/chain/storage/tiered_storage.cpp
@@ -0,0 +1,581 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one + * or more contributor license agreements. See the NOTICE file + * distributed with this work for additional information + * regarding copyright ownership. The ASF licenses this file + * to you 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 "chain/storage/tiered_storage.h" + +#include <chrono> +#include <google/protobuf/util/json_util.h> +#include <glog/logging.h> + +#include "chain/storage/in_memory_index.h" +#include "chain/storage/leveldb.h" + +namespace resdb { +namespace storage { + +namespace { + +constexpr const char* kManifestKey = "_tiered_manifest"; +constexpr const char* kColdThresholdKey = "_cold_threshold"; +constexpr int kDefaultColdThreshold = 2; +constexpr int kDefaultPollIntervalSeconds = 60; +constexpr int kDefaultBatchSize = 100; +constexpr int kDefaultCheckpointWatermark = 5; + +} + +TieredStorage::TieredStorage( + std::unique_ptr<Storage> hot_storage, + std::unique_ptr<Storage> warm_storage, + std::unique_ptr<IPFSClient> cold_client, + const TieredStorageConfig& config) + : hot_storage_(std::move(hot_storage)), + warm_storage_(std::move(warm_storage)), + cold_client_(std::move(cold_client)), + config_(config), + manifest_key_(kManifestKey) { + if (!config_.enabled()) { + LOG(WARNING) << "TieredStorage created but tiering is disabled"; + } else { + LOG(INFO) << "TieredStorage enabled with cold_threshold = " + << config_.cold_threshold_checkpoint(); + } + + if (config_.cold_threshold_checkpoint() < 0) { + config_.set_cold_threshold_checkpoint(kDefaultColdThreshold); + } + + if (config_.checkpoint_watermark() <= 0) { + config_.set_checkpoint_watermark(kDefaultCheckpointWatermark); + } + + if (config_.enabled() && cold_client_ && cold_client_->IsEnabled()) { + index_ = std::make_unique<InMemoryHashIndex>(); + LoadManifestFromStorage(warm_storage_.get()); + StartMigration(); + } +} + +TieredStorage::~TieredStorage() { + StopMigration(); +} + +int TieredStorage::SetValue(const std::string& key, const std::string& value) { + return SetValueInternal(hot_storage_.get(), key, value); +} + +int TieredStorage::SetValueWithSeq(const std::string& key, + const std::string& value, uint64_t seq) { + int ret = hot_storage_->SetValueWithSeq(key, value, seq); + if (ret == 0) { + uint64_t prev = max_seq_.load(); + while (seq > prev && !max_seq_.compare_exchange_weak(prev, seq)) { + } + } + return ret; +} + +std::string TieredStorage::GetValue(const std::string& key) { + return GetValueWithFallback(key); +} + +std::pair<std::string, uint64_t> TieredStorage::GetValueWithSeq( + const std::string& key, uint64_t seq) { + auto hot_result = hot_storage_->GetValueWithSeq(key, seq); + if (!hot_result.first.empty()) { + LOG(INFO) << "TieredStorage: found value in hot storage for key: " << key; + return hot_result; + } + + auto warm_result = warm_storage_->GetValueWithSeq(key, seq); + if (!warm_result.first.empty()) { + LOG(INFO) << "TieredStorage: found value in warm storage for key: " << key; + return warm_result; + } + + if (IsColdEnabled()) { + std::string cold_value = GetValueFromCold(key); + if (!cold_value.empty()) { + LOG(INFO) << "TieredStorage: found value in cold storage for key: " << key; + return {cold_value, 0}; + } + } + + LOG(INFO) << "TieredStorage: key not found: " << key; + return {"", 0}; +} + +std::string TieredStorage::GetRange(const std::string& min_key, + const std::string& max_key) { + std::string result = warm_storage_->GetRange(min_key, max_key); + if (!result.empty()) { + return result; + } + + return result; +} + +int TieredStorage::SetValueWithVersion(const std::string& key, + const std::string& value, int version) { + return hot_storage_->SetValueWithVersion(key, value, version); +} + +std::pair<std::string, int> TieredStorage::GetValueWithVersion( + const std::string& key, int version) { + auto result = hot_storage_->GetValueWithVersion(key, version); + if (!result.first.empty()) { + return result; + } + + return warm_storage_->GetValueWithVersion(key, version); +} + +std::map<std::string, std::vector<std::pair<std::string, uint64_t>>> +TieredStorage::GetAllItemsWithSeq() { + return warm_storage_->GetAllItemsWithSeq(); +} + +std::map<std::string, std::pair<std::string, int>> TieredStorage::GetAllItems() { + return warm_storage_->GetAllItems(); +} + +std::map<std::string, std::pair<std::string, int>> TieredStorage::GetKeyRange( + const std::string& min_key, const std::string& max_key) { + return warm_storage_->GetKeyRange(min_key, max_key); +} + +std::vector<std::pair<std::string, int>> TieredStorage::GetHistory( + const std::string& key, int min_version, int max_version) { + auto result = hot_storage_->GetHistory(key, min_version, max_version); + if (!result.empty()) { + return result; + } + + return warm_storage_->GetHistory(key, min_version, max_version); +} + +std::vector<std::pair<std::string, int>> TieredStorage::GetTopHistory( + const std::string& key, int number) { + auto result = hot_storage_->GetTopHistory(key, number); + if (!result.empty()) { + return result; + } + + return warm_storage_->GetTopHistory(key, number); +} + +bool TieredStorage::Flush() { + hot_storage_->Flush(); + warm_storage_->Flush(); + return true; +} + +uint64_t TieredStorage::GetLastCheckpoint() { + if (config_.hot_backend() == TieredStorageConfig::LEVELDB && hot_storage_) { + return hot_storage_->GetLastCheckpoint(); + } + return max_seq_.load(); +} + +void TieredStorage::SetColdThreshold(int checkpoint_threshold) { + config_.set_cold_threshold_checkpoint(checkpoint_threshold); + LOG(INFO) << "Set cold threshold to " << checkpoint_threshold; +} + +int TieredStorage::GetColdThreshold() const { + return config_.cold_threshold_checkpoint(); +} + +bool TieredStorage::IsColdData(const std::string& key) const { + return !GetIndexCID(key).empty(); +} + +bool TieredStorage::IsColdEnabled() const { + return config_.enabled() && cold_client_ && cold_client_->IsEnabled(); +} + +std::string TieredStorage::GetCID(const std::string& key) const { + return GetIndexCID(key); +} + +std::unique_ptr<Storage> TieredStorage::Create( + std::unique_ptr<Storage> hot_storage, + std::unique_ptr<Storage> warm_storage, + const IPFSConfig& ipfs_config, + const TieredStorageConfig& tiered_config) { + if (!tiered_config.enabled()) { + LOG(INFO) << "Tiered storage disabled, using warm storage only"; + return warm_storage; + } + + auto cold_client = IPFSClient::Create(ipfs_config); + if (!cold_client->IsEnabled()) { + LOG(WARNING) << "IPFS client not enabled, falling back to warm storage"; + return warm_storage; + } + + return std::make_unique<TieredStorage>( + std::move(hot_storage), + std::move(warm_storage), + std::move(cold_client), + tiered_config); +} + +std::string TieredStorage::GetValueWithFallback(const std::string& key) { + std::string value = GetValueInternal(hot_storage_.get(), key); + if (!value.empty()) { + LOG(INFO) << "TieredStorage: found value in hot storage for key: " << key; + return value; + } + + value = GetValueInternal(warm_storage_.get(), key); + if (!value.empty()) { + LOG(INFO) << "TieredStorage: found value in warm storage for key: " << key; + return value; + } + + if (IsColdEnabled()) { + value = GetValueFromCold(key); + if (!value.empty()) { + LOG(INFO) << "TieredStorage: found value in cold storage for key: " << key; + return value; + } + } + + LOG(INFO) << "TieredStorage: key not found: " << key; + return ""; +} + +std::string TieredStorage::GetValueFromCold(const std::string& key) { + std::string cid = GetIndexCID(key); + if (cid.empty()) { + LOG(INFO) << "TieredStorage: no CID found for key: " << key; + return ""; + } + + auto start = std::chrono::high_resolution_clock::now(); + std::string result = cold_client_->Cat(cid); + auto end = std::chrono::high_resolution_clock::now(); + auto us = std::chrono::duration_cast<std::chrono::microseconds>(end - start).count(); + LOG(INFO) << "[timer] cold_read key=" << key << " cid=" << cid << " us=" << us; + return result; +} + +std::string TieredStorage::GetIndexCID(const std::string& key) const { + if (index_) { + auto start = std::chrono::high_resolution_clock::now(); + std::string cid = index_->Get(key); + auto end = std::chrono::high_resolution_clock::now(); + auto us = std::chrono::duration_cast<std::chrono::microseconds>(end - start).count(); + LOG(INFO) << "[timer] index_lookup key=" << key << " us=" << us; + return cid; + } + return ""; +} + +int TieredStorage::SetValueInternal(Storage* storage, + const std::string& key, + const std::string& value) { + if (!storage) { + LOG(ERROR) << "TieredStorage: storage is null"; + return -1; + } + return storage->SetValue(key, value); +} + +std::string TieredStorage::GetValueInternal(Storage* storage, + const std::string& key) { + if (!storage) { + LOG(ERROR) << "TieredStorage: storage is null"; + return ""; + } + return storage->GetValue(key); +} + +bool TieredStorage::LoadManifest() { + return LoadManifestFromStorage(warm_storage_.get()); +} + +bool TieredStorage::SaveManifest() { + return SaveManifestToStorage(warm_storage_.get()); +} + +bool TieredStorage::AddToIndex(const std::string& key, const std::string& cid, + uint64_t min_seq, uint64_t max_seq) { + if (!IsColdEnabled() || !index_) { + LOG(WARNING) << "TieredStorage: cold storage not enabled or no index"; + return false; + } + + index_->Add(key, cid); + + auto* mapping = manifest_.add_range_mappings(); + mapping->set_start_key(key); + mapping->set_end_key(key); + mapping->set_ipfs_cid(cid); + mapping->set_min_checkpoint(min_seq); + mapping->set_max_checkpoint(max_seq); + + manifest_.set_total_keys(manifest_.total_keys() + 1); + manifest_.set_cold_keys(manifest_.cold_keys() + 1); + manifest_.set_last_updated_timestamp(time(nullptr)); + + LOG(INFO) << "TieredStorage: added to index, key=" << key + << ", cid=" << cid << ", range=[" << min_seq << "," << max_seq << "]"; + + return SaveManifestToStorage(warm_storage_.get()); +} + +bool TieredStorage::LoadManifestFromStorage(Storage* storage) { + if (!storage) { + LOG(ERROR) << "TieredStorage: cannot load manifest, storage is null"; + return false; + } + + std::string manifest_data = storage->GetValue(manifest_key_); + if (manifest_data.empty()) { + LOG(INFO) << "TieredStorage: no manifest found, starting fresh"; + manifest_.set_version(1); + manifest_.set_total_keys(0); + manifest_.set_cold_keys(0); + manifest_.set_last_updated_timestamp(time(nullptr)); + return true; + } + + if (!manifest_.ParseFromString(manifest_data)) { + LOG(ERROR) << "TieredStorage: failed to parse manifest"; + return false; + } + + if (index_) { + index_->Clear(); + for (const auto& mapping : manifest_.range_mappings()) { + index_->Add(mapping.start_key(), mapping.ipfs_cid()); + } + } + + LOG(INFO) << "TieredStorage: loaded manifest, " + << "total_keys=" << manifest_.total_keys() + << ", cold_keys=" << manifest_.cold_keys(); + manifest_loaded_ = true; + return true; +} + +bool TieredStorage::SaveManifestToStorage(Storage* storage) { + if (!storage) { + LOG(ERROR) << "TieredStorage: cannot save manifest, storage is null"; + return false; + } + + std::string manifest_data; + if (!manifest_.SerializeToString(&manifest_data)) { + LOG(ERROR) << "TieredStorage: failed to serialize manifest"; + return false; + } + + int ret = storage->SetValue(manifest_key_, manifest_data); + if (ret != 0) { + LOG(ERROR) << "TieredStorage: failed to save manifest, ret=" << ret; + return false; + } + + LOG(INFO) << "TieredStorage: saved manifest, " + << "size=" << manifest_data.size() << " bytes"; + return true; +} + +void TieredStorage::StartMigration() { + if (!IsColdEnabled()) { + LOG(WARNING) << "TieredStorage: cannot start migration, cold storage not enabled"; + return; + } + + if (migration_running_.load()) { + LOG(WARNING) << "TieredStorage: migration thread already running"; + return; + } + + if (!config_.auto_migration_enabled()) { + LOG(INFO) << "TieredStorage: auto migration not enabled in config"; + return; + } + + migration_running_.store(true); + migration_thread_ = std::thread(&TieredStorage::MigrationLoop, this); + LOG(INFO) << "TieredStorage: migration thread started"; +} + +void TieredStorage::StopMigration() { + if (!migration_running_.load()) { + return; + } + + migration_running_.store(false); + if (migration_thread_.joinable()) { + migration_thread_.join(); + } + LOG(INFO) << "TieredStorage: migration thread stopped, " + << "total_migrated=" << migrated_keys_.load(); +} + +void TieredStorage::MigrationLoop() { + int poll_interval = config_.poll_interval_seconds(); + if (poll_interval <= 0) { + poll_interval = kDefaultPollIntervalSeconds; + } + + LOG(INFO) << "TieredStorage: migration polling started, interval=" + << poll_interval << "s"; + + while (migration_running_.load()) { + try { + MigrateColdData(); + } catch (const std::exception& e) { + LOG(ERROR) << "TieredStorage: migration error: " << e.what(); + } + + for (int i = 0; i < poll_interval && migration_running_.load(); ++i) { + std::this_thread::sleep_for(std::chrono::seconds(1)); + } + } + + LOG(INFO) << "TieredStorage: migration polling stopped"; +} + +bool TieredStorage::MigrateColdData() { + auto cycle_start = std::chrono::high_resolution_clock::now(); + uint64_t current_checkpoint = GetLastCheckpoint(); + int watermark = config_.checkpoint_watermark(); + if (watermark <= 0) { + watermark = kDefaultCheckpointWatermark; + } + uint64_t threshold = config_.cold_threshold_checkpoint() * watermark; + + if (current_checkpoint <= threshold) { + LOG(INFO) << "TieredStorage: checkpoint " << current_checkpoint + << " not past threshold " << threshold; + return false; + } + + uint64_t cold_threshold_seq = current_checkpoint - threshold; + + LOG(INFO) << "TieredStorage: migrating data with seq < " << cold_threshold_seq; + + int batch_size = config_.batch_size(); + if (batch_size <= 0) { + batch_size = kDefaultBatchSize; + } + + int migrated = 0; + + auto all_items = hot_storage_->GetAllItemsWithSeq(); + + for (const auto& [key, value_list] : all_items) { + for (const auto& [value, seq] : value_list) { + if (seq > cold_threshold_seq) { + continue; + } + + if (MigrateKey(key, value, seq)) { + migrated++; + } + + if (migrated >= batch_size) { + LOG(INFO) << "TieredStorage: batch size reached: " << migrated; + break; + } + } + + if (migrated >= batch_size) { + break; + } + } + + auto cycle_end = std::chrono::high_resolution_clock::now(); + auto cycle_us = std::chrono::duration_cast<std::chrono::microseconds>(cycle_end - cycle_start).count(); + LOG(INFO) << "[timer] migrate_cycle keys=" << migrated << " us=" << cycle_us; + + return migrated > 0; +} + +bool TieredStorage::MigrateKey(const std::string& key, const std::string& value, uint64_t seq) { + std::string old_cid = GetIndexCID(key); + + auto add_start = std::chrono::high_resolution_clock::now(); + std::string cid = cold_client_->Add(value); + auto add_end = std::chrono::high_resolution_clock::now(); + auto add_us = std::chrono::duration_cast<std::chrono::microseconds>(add_end - add_start).count(); + LOG(INFO) << "[timer] ipfs_add key=" << key << " us=" << add_us; + if (cid.empty()) { + LOG(ERROR) << "TieredStorage: failed to upload to IPFS: key=" << key; + return false; + } + + LOG(INFO) << "TieredStorage: uploaded to IPFS: key=" << key << ", cid=" << cid; + + if (!AddToIndex(key, cid, seq, seq)) { + LOG(ERROR) << "TieredStorage: failed to add to index: key=" << key; + return false; + } + + if (DeleteKeyFromHot(key)) { + LOG(INFO) << "TieredStorage: evicted from hot storage: key=" << key; + } + + if (!old_cid.empty() && old_cid != cid) { + if (cold_client_->Unpin(old_cid)) { + LOG(INFO) << "TieredStorage: unpinned stale CID=" << old_cid + << " for key=" << key; + } else { + LOG(WARNING) << "TieredStorage: failed to unpin stale CID=" << old_cid + << " for key=" << key; + } + } + + migrated_keys_.fetch_add(1); + LOG(INFO) << "TieredStorage: key migrated successfully: key=" << key; + return true; +} + +bool TieredStorage::DeleteKeyFromWarm(const std::string& key) { + if (!IsTiered() || !IsColdEnabled()) { + LOG(ERROR) << "TieredStorage: delete attempted but tiered storage not enabled"; + return false; + } + + auto* deletable = dynamic_cast<DeletableStorage*>(warm_storage_.get()); + if (!deletable) { + LOG(ERROR) << "TieredStorage: warm storage does not support deletion"; + return false; + } + + return deletable->DeleteKey(key); +} + +bool TieredStorage::DeleteKeyFromHot(const std::string& key) { + auto* deletable = dynamic_cast<DeletableStorage*>(hot_storage_.get()); + if (!deletable) { + LOG(WARNING) << "TieredStorage: hot storage does not support deletion"; + return false; + } + return deletable->DeleteKey(key); +} + +} // namespace storage +} // namespace resdb \ No newline at end of file
diff --git a/chain/storage/tiered_storage.h b/chain/storage/tiered_storage.h new file mode 100644 index 0000000..2d07a92 --- /dev/null +++ b/chain/storage/tiered_storage.h
@@ -0,0 +1,133 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one + * or more contributor license agreements. See the NOTICE file + * distributed with this work for additional information + * regarding copyright ownership. The ASF licenses this file + * to you 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 <atomic> +#include <memory> +#include <string> +#include <thread> + +#include "chain/storage/interface/secondary_index.h" +#include "chain/storage/storage.h" +#include "chain/storage/ipfs_client.h" +#include "chain/storage/proto/ipfs_config.pb.h" +#include "chain/storage/proto/tiered_index.pb.h" + +namespace resdb { +namespace storage { + +class TieredStorage : public Storage { + public: + TieredStorage(std::unique_ptr<Storage> hot_storage, + std::unique_ptr<Storage> warm_storage, + std::unique_ptr<IPFSClient> cold_client, + const TieredStorageConfig& config); + ~TieredStorage() override; + + int SetValue(const std::string& key, const std::string& value) override; + int SetValueWithSeq(const std::string& key, const std::string& value, + uint64_t seq) override; + std::string GetValue(const std::string& key) override; + std::pair<std::string, uint64_t> GetValueWithSeq(const std::string& key, + uint64_t seq) override; + std::string GetRange(const std::string& min_key, + const std::string& max_key) override; + + int SetValueWithVersion(const std::string& key, const std::string& value, + int version) override; + std::pair<std::string, int> GetValueWithVersion(const std::string& key, + int version) override; + + std::map<std::string, std::vector<std::pair<std::string, uint64_t>>> + GetAllItemsWithSeq() override; + std::map<std::string, std::pair<std::string, int>> GetAllItems() override; + std::map<std::string, std::pair<std::string, int>> GetKeyRange( + const std::string& min_key, const std::string& max_key) override; + + std::vector<std::pair<std::string, int>> GetHistory(const std::string& key, + int min_version, + int max_version) override; + std::vector<std::pair<std::string, int>> GetTopHistory(const std::string& key, + int number) override; + + bool Flush() override; + uint64_t GetLastCheckpoint() override; + + void SetColdThreshold(int checkpoint_threshold); + int GetColdThreshold() const; + bool IsColdData(const std::string& key) const; + bool IsColdEnabled() const; + std::string GetCID(const std::string& key) const; + bool IsTiered() const { return config_.enabled(); } + + static std::unique_ptr<Storage> Create( + std::unique_ptr<Storage> hot_storage, + std::unique_ptr<Storage> warm_storage, + const IPFSConfig& ipfs_config, + const TieredStorageConfig& tiered_config); + + bool LoadManifest(); + bool SaveManifest(); + bool AddToIndex(const std::string& key, const std::string& cid, + uint64_t min_seq, uint64_t max_seq); + uint64_t GetColdKeyCount() const { return manifest_.cold_keys(); } + + void StartMigration(); + void StopMigration(); + uint64_t GetMigratedKeyCount() const { return migrated_keys_.load(); } + + bool LoadManifestFromStorage(Storage* storage); + + private: + std::string GetValueWithFallback(const std::string& key); + std::string GetValueFromCold(const std::string& key); + std::string GetIndexCID(const std::string& key) const; + + int SetValueInternal(Storage* storage, const std::string& key, + const std::string& value); + std::string GetValueInternal(Storage* storage, const std::string& key); + bool SaveManifestToStorage(Storage* storage); + + void MigrationLoop(); + bool MigrateColdData(); + bool MigrateKey(const std::string& key, const std::string& value, uint64_t seq); + bool DeleteKeyFromWarm(const std::string& key); + bool DeleteKeyFromHot(const std::string& key); + + std::unique_ptr<Storage> hot_storage_; + std::unique_ptr<Storage> warm_storage_; + std::unique_ptr<IPFSClient> cold_client_; + TieredStorageConfig config_; + + IndexManifest manifest_; + std::string manifest_key_; + bool manifest_loaded_ = false; + + std::unique_ptr<SecondaryIndex> index_; + std::atomic<uint64_t> max_seq_{0}; + + std::thread migration_thread_; + std::atomic<bool> migration_running_{false}; + std::atomic<uint64_t> migrated_keys_{0}; + +}; + +} // namespace storage +} // namespace resdb \ No newline at end of file
diff --git a/chain/storage/tiered_storage_test.cpp b/chain/storage/tiered_storage_test.cpp new file mode 100644 index 0000000..36026f1 --- /dev/null +++ b/chain/storage/tiered_storage_test.cpp
@@ -0,0 +1,217 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one + * or more contributor license agreements. See the NOTICE file + * distributed with this work for additional information + * regarding copyright ownership. The ASF licenses this file + * to you 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 <glog/logging.h> +#include <gtest/gtest.h> + +#include <filesystem> + +#include "chain/storage/leveldb.h" +#include "chain/storage/memory_db.h" +#include "chain/storage/ipfs_client.h" +#include "chain/storage/tiered_storage.h" + +namespace resdb { +namespace storage { +namespace { + +class StubIPFSClient : public IPFSClient { + public: + StubIPFSClient(bool enabled) : enabled_(enabled) {} + + std::string Add(const std::string& data) override { + return data.empty() ? "" : "stub_cid_" + data.substr(0, 10); + } + std::string AddDAG(const std::string& json_data) override { + return "stub_dag_cid"; + } + std::string Cat(const std::string& cid) override { + return "fetched_" + cid; + } + std::string GetDAG(const std::string& cid) override { + return "dag_" + cid; + } + bool Exists(const std::string& cid) override { + return !cid.empty(); + } + bool Unpin(const std::string& cid) override { + (void)cid; + return true; + } + bool IsEnabled() const override { + return enabled_; + } + + private: + bool enabled_; +}; + +class TieredStorageTest : public ::testing::Test { + protected: + void SetUp() override { + std::filesystem::remove_all(leveldb_path_.c_str()); + } + + void TearDown() override { + std::filesystem::remove_all(leveldb_path_.c_str()); + } + + std::string leveldb_path_ = "/tmp/tiered_storage_test"; +}; + +TEST_F(TieredStorageTest, CreateDisabled) { + TieredStorageConfig config; + config.set_enabled(false); + + auto warm = NewResLevelDB(leveldb_path_); + auto hot = NewMemoryDB(); + std::unique_ptr<IPFSClient> ipfs(new StubIPFSClient(false)); + + auto storage = std::make_unique<TieredStorage>( + std::move(hot), std::move(warm), std::move(ipfs), config); + + EXPECT_FALSE(storage->IsTiered()); +} + +TEST_F(TieredStorageTest, CreateEnabled) { + TieredStorageConfig config; + config.set_enabled(true); + config.set_cold_threshold_checkpoint(2); + + auto warm = NewResLevelDB(leveldb_path_); + auto hot = NewMemoryDB(); + std::unique_ptr<IPFSClient> ipfs(new StubIPFSClient(true)); + + auto storage = std::make_unique<TieredStorage>( + std::move(hot), std::move(warm), std::move(ipfs), config); + + EXPECT_TRUE(storage->IsTiered()); + EXPECT_TRUE(storage->IsColdEnabled()); +} + +TEST_F(TieredStorageTest, SetValueHotStorage) { + TieredStorageConfig config; + config.set_enabled(false); + + auto warm = NewResLevelDB(leveldb_path_); + auto hot = NewMemoryDB(); + std::unique_ptr<IPFSClient> ipfs(new StubIPFSClient(false)); + + auto storage = std::make_unique<TieredStorage>( + std::move(hot), std::move(warm), std::move(ipfs), config); + + EXPECT_EQ(storage->SetValue("test_key", "test_value"), 0); + EXPECT_EQ(storage->GetValue("test_key"), "test_value"); +} + +TEST_F(TieredStorageTest, FallbackToWarmStorage) { + TieredStorageConfig config; + config.set_enabled(false); + + auto warm = NewResLevelDB(leveldb_path_); + auto hot = NewMemoryDB(); + warm->SetValue("warm_key", "warm_value"); + std::unique_ptr<IPFSClient> ipfs(new StubIPFSClient(false)); + + auto storage = std::make_unique<TieredStorage>( + std::move(hot), std::move(warm), std::move(ipfs), config); + + EXPECT_EQ(storage->GetValue("warm_key"), "warm_value"); +} + +TEST_F(TieredStorageTest, GetColdThreshold) { + TieredStorageConfig config; + config.set_enabled(true); + config.set_cold_threshold_checkpoint(5); + + auto warm = NewResLevelDB(leveldb_path_); + auto hot = NewMemoryDB(); + std::unique_ptr<IPFSClient> ipfs(new StubIPFSClient(true)); + + auto storage = std::make_unique<TieredStorage>( + std::move(hot), std::move(warm), std::move(ipfs), config); + + EXPECT_EQ(storage->GetColdThreshold(), 5); + storage->SetColdThreshold(10); + EXPECT_EQ(storage->GetColdThreshold(), 10); +} + +TEST_F(TieredStorageTest, GetLastCheckpointTracksMaxSeq) { + TieredStorageConfig config; + config.set_enabled(false); + + auto warm = NewResLevelDB(leveldb_path_); + auto hot = NewMemoryDB(); + std::unique_ptr<IPFSClient> ipfs(new StubIPFSClient(false)); + + auto storage = std::make_unique<TieredStorage>( + std::move(hot), std::move(warm), std::move(ipfs), config); + + storage->SetValueWithSeq("key1", "value1", 100); + storage->SetValueWithSeq("key2", "value2", 50); + + EXPECT_EQ(storage->GetLastCheckpoint(), 100); +} + +TEST_F(TieredStorageTest, GetValueWithSeq) { + TieredStorageConfig config; + config.set_enabled(false); + + auto warm = NewResLevelDB(leveldb_path_); + auto hot = NewMemoryDB(); + auto* warm_ptr = warm.get(); + auto* hot_ptr = hot.get(); + std::unique_ptr<IPFSClient> ipfs(new StubIPFSClient(false)); + + auto storage = std::make_unique<TieredStorage>( + std::move(hot), std::move(warm), std::move(ipfs), config); + + hot_ptr->SetValueWithSeq("key1", "value1", 1); + warm_ptr->SetValueWithSeq("key2", "value2", 2); + + auto result1 = storage->GetValueWithSeq("key1", 0); + EXPECT_EQ(result1.first, "value1"); + EXPECT_EQ(result1.second, 1); + + auto result2 = storage->GetValueWithSeq("key2", 0); + EXPECT_EQ(result2.first, "value2"); + EXPECT_EQ(result2.second, 2); +} + +TEST_F(TieredStorageTest, SetValueWithVersion) { + TieredStorageConfig config; + config.set_enabled(false); + + auto warm = NewResLevelDB(leveldb_path_); + auto hot = NewMemoryDB(); + std::unique_ptr<IPFSClient> ipfs(new StubIPFSClient(false)); + + auto storage = std::make_unique<TieredStorage>( + std::move(hot), std::move(warm), std::move(ipfs), config); + + EXPECT_EQ(storage->SetValueWithVersion("key1", "value1", 0), 0); + + auto result = storage->GetValueWithVersion("key1", 1); + EXPECT_EQ(result.first, "value1"); + EXPECT_EQ(result.second, 1); +} + +} // namespace +} // namespace storage +} // namespace resdb \ No newline at end of file
diff --git a/common/lru/lru_cache.cpp b/common/lru/lru_cache.cpp index 75033d4..23b4fff 100644 --- a/common/lru/lru_cache.cpp +++ b/common/lru/lru_cache.cpp
@@ -87,6 +87,16 @@ } template <typename KeyType, typename ValueType> +void LRUCache<KeyType, ValueType>::Remove(KeyType key) { + auto it = lookup_.find(key); + if (it != lookup_.end()) { + key_list_.erase(rlookup_[key]); + lookup_.erase(it); + rlookup_.erase(key); + } +} + +template <typename KeyType, typename ValueType> void LRUCache<KeyType, ValueType>::Flush() { lookup_.clear(); key_list_.clear();
diff --git a/common/lru/lru_cache.h b/common/lru/lru_cache.h index aa4a17d..de53c35 100644 --- a/common/lru/lru_cache.h +++ b/common/lru/lru_cache.h
@@ -32,6 +32,7 @@ void Put(KeyType key, ValueType value); int GetCapacity(); void SetCapacity(int new_capacity); + void Remove(KeyType key); void Flush(); int GetCacheHits() const; int GetCacheMisses() const;
diff --git a/docker/Makefile b/docker/Makefile new file mode 100644 index 0000000..d6528b1 --- /dev/null +++ b/docker/Makefile
@@ -0,0 +1,66 @@ +.PHONY: help start stop restart logs status benchmark metrics clean baseline tiered + +RESDB_IMAGE ?= resilientdb:latest +IPFS_IMAGE ?= ipfs/kubo:latest + +help: + @echo "ResilientDB Docker Cluster Management" + @echo "" + @echo "Usage:" + @echo " make start Start 4-node cluster (LEVELDB_ONLY mode)" + @echo " make baseline Same as make start" + @echo " make tiered Start 4-node cluster with tiered storage" + @echo " make stop Stop all containers" + @echo " make restart Restart cluster" + @echo " make logs Show logs" + @echo " make status Show container status" + @echo " make benchmark Run benchmark (TPS=100, 60s)" + @echo " make metrics Collect metrics" + @echo " make clean Clean up containers and volumes" + @echo "" + @echo "Environment Variables:" + @echo " RESDB_IMAGE ResilientDB image (default: resilientdb:latest)" + @echo " IPFS_IMAGE IPFS image (default: ipfs/kubo:latest)" + @echo " TPS Benchmark TPS (default: 100)" + @echo " DURATION Benchmark duration in seconds (default: 60)" + @echo "" + @echo "Examples:" + @echo " make start RESDB_IMAGE=myregistry/resilientdb:v1.0" + @echo " make tiered COLD_THRESHOLD=3" + @echo " make benchmark TPS=500 DURATION=120" + +start baseline: + @echo "Starting cluster in baseline mode..." + STORAGE_BACKEND=LEVELDB_ONLY TIERED_ENABLED=false docker-compose up -d + @echo "Cluster started. Use 'make logs' to view logs." + +tiered: + @echo "Starting cluster with tiered storage..." + STORAGE_BACKEND=TIERED TIERED_ENABLED=true docker-compose up -d + @echo "Cluster started with tiered storage. Use 'make logs' to view logs." + +stop: + @echo "Stopping cluster..." + docker-compose down + +restart: stop start + +logs: + docker-compose logs -f + +status: + docker-compose ps + +benchmark: + @./run_benchmark.sh $(TPS) $(DURATION) 1024 localhost 18001 + +metrics: + @./collect_metrics.sh metrics.json 4 + +clean: + @echo "Cleaning up..." + docker-compose down -v + @echo "Done. Use 'make start' to restart." + +validate-config: + @docker-compose config > /dev/null && echo "Config is valid" || echo "Config is invalid" \ No newline at end of file
diff --git a/docker/collect_metrics.sh b/docker/collect_metrics.sh new file mode 100755 index 0000000..14880eb --- /dev/null +++ b/docker/collect_metrics.sh
@@ -0,0 +1,78 @@ +#!/bin/bash + +set -e + +OUTPUT="${1:-metrics.json}" +NODES="${2:-4}" + +echo "==========================================" +echo "Collecting Metrics" +echo "==========================================" +echo "Output: $OUTPUT" +echo "Nodes: $NODES" +echo "==========================================" + +METRICS="{ + \"timestamp\": \"$(date -Iseconds)\", + \"nodes\": [" + +for i in $(seq 0 $((NODES - 1)); do + NODE_PORT=$((18000 + i * 10)) + + echo "Collecting metrics from node $i (port $NODE_PORT)..." + + NODE_METRICS=$(curl -s "http://localhost:${NODE_PORT}/metrics" 2>/dev/null || echo "") + + TX_COUNT=$(echo "$NODE_METRICS" | grep "resdb_transactions_total" | awk '{print $2}' || echo "0") + TX_LATENCY_P50=$(echo "$NODE_METRICS" | grep "resdb_tx_latency_bucket{le=\"1\"" | awk '{print $2}' || echo "0") + CHECKPOINT=$(echo "$NODE_METRICS" | grep "resdb_last_checkpoint" | awk '{print $2}' || echo "0") + + if [ $i -gt 0 ]; then + METRICS+="," + fi + + METRICS+=" + { + \"node_id\": $i, + \"port\": $NODE_PORT, + \"tx_count\": $TX_COUNT, + \"tx_latency_p50\": $TX_LATENCY_P50, + \"checkpoint\": $CHECKPOINT + }" +done + +METRICS+=" + ], + \"ipfs\": [" + +for i in $(seq 0 $((NODES - 1))); do + IPFS_PORT=$((5001 + i * 100)) + + echo "Collecting IPFS metrics from node $i (port $IPFS_PORT)..." + + IPFS_STATS=$(curl -s "http://localhost:${IPFS_PORT}/api/v0/repo/stat" 2>/dev/null || echo "{}") + + if [ $i -gt 0 ]; then + METRICS+="," + fi + + METRICS+=" + { + \"node_id\": $i, + \"port\": $IPFS_PORT, + \"stats\": $IPFS_STATS + }" +done + +METRICS+=" + ] +}" + +echo "$METRICS" > "$OUTPUT" + +echo "" +echo "==========================================" +echo "Metrics collected!" +echo "==========================================" +echo "Output: $OUTPUT" +echo "==========================================" \ No newline at end of file
diff --git a/docker/configs/prometheus/prometheus.yml b/docker/configs/prometheus/prometheus.yml new file mode 100644 index 0000000..107b91d --- /dev/null +++ b/docker/configs/prometheus/prometheus.yml
@@ -0,0 +1,20 @@ +global: + scrape_interval: 15s + evaluation_interval: 15s + +scrape_configs: + - job_name: 'resilientdb' + static_configs: + - targets: + - 'resilientdb-0:18000' + - 'resilientdb-1:18010' + - 'resilientdb-2:18020' + - 'resilientdb-3:18030' + + - job_name: 'ipfs' + static_configs: + - targets: + - 'ipfs-0:5001' + - 'ipfs-1:5001' + - 'ipfs-2:5001' + - 'ipfs-3:5001' \ No newline at end of file
diff --git a/docker/docker-compose.yml b/docker/docker-compose.yml new file mode 100644 index 0000000..2b1737c --- /dev/null +++ b/docker/docker-compose.yml
@@ -0,0 +1,185 @@ +version: '3.8' + +services: + resilientdb-0: + image: ${RESDB_IMAGE:-resilientdb:latest} + hostname: resilientdb-0 + container_name: resilientdb-0 + ports: + - "18001:18001" + - "18000:18000" + volumes: + - ./configs/node0:/app/config + - leveldb0:/app/data + environment: + - NODE_ID=0 + - NODE_HOST=resilientdb-0 + - STORAGE_BACKEND=${STORAGE_BACKEND:-LEVELDB_ONLY} + - IPFS_ENDPOINT=http://ipfs-0:5001 + - TIERED_ENABLED=${TIERED_ENABLED:-false} + - COLD_THRESHOLD=${COLD_THRESHOLD:-2} + networks: + - resdb-net + depends_on: + - ipfs-0 + deploy: + resources: + limits: + memory: 2G + reservations: + memory: 1G + + ipfs-0: + image: ${IPFS_IMAGE:-ipfs/kubo:latest} + hostname: ipfs-0 + container_name: ipfs-0 + volumes: + - ipfs0:/data/ipfs + environment: + - IPFS_SWARM_ADDRS=/ip4/0.0.0.0/tcp/4001 + ports: + - "5001:5001" + - "8080:8080" + networks: + - resdb-net + + resilientdb-1: + image: ${RESDB_IMAGE:-resilientdb:latest} + hostname: resilientdb-1 + container_name: resilientdb-1 + ports: + - "18011:18001" + - "18010:18000" + volumes: + - ./configs/node1:/app/config + - leveldb1:/app/data + environment: + - NODE_ID=1 + - NODE_HOST=resilientdb-1 + - STORAGE_BACKEND=${STORAGE_BACKEND:-LEVELDB_ONLY} + - IPFS_ENDPOINT=http://ipfs-1:5001 + - TIERED_ENABLED=${TIERED_ENABLED:-false} + - COLD_THRESHOLD=${COLD_THRESHOLD:-2} + networks: + - resdb-net + depends_on: + - ipfs-1 + + ipfs-1: + image: ${IPFS_IMAGE:-ipfs/kubo:latest} + hostname: ipfs-1 + container_name: ipfs-1 + volumes: + - ipfs1:/data/ipfs + environment: + - IPFS_SWARM_ADDRS=/ip4/0.0.0.0/tcp/4001 + ports: + - "5011:5001" + - "8081:8080" + networks: + - resdb-net + + resilientdb-2: + image: ${RESDB_IMAGE:-resilientdb:latest} + hostname: resilientdb-2 + container_name: resilientdb-2 + ports: + - "18021:18001" + - "18020:18000" + volumes: + - ./configs/node2:/app/config + - leveldb2:/app/data + environment: + - NODE_ID=2 + - NODE_HOST=resilientdb-2 + - STORAGE_BACKEND=${STORAGE_BACKEND:-LEVELDB_ONLY} + - IPFS_ENDPOINT=http://ipfs-2:5001 + - TIERED_ENABLED=${TIERED_ENABLED:-false} + - COLD_THRESHOLD=${COLD_THRESHOLD:-2} + networks: + - resdb-net + depends_on: + - ipfs-2 + + ipfs-2: + image: ${IPFS_IMAGE:-ipfs/kubo:latest} + hostname: ipfs-2 + container_name: ipfs-2 + volumes: + - ipfs2:/data/ipfs + environment: + - IPFS_SWARM_ADDRS=/ip4/0.0.0.0/tcp/4001 + ports: + - "5021:5001" + - "8082:8080" + networks: + - resdb-net + + resilientdb-3: + image: ${RESDB_IMAGE:-resilientdb:latest} + hostname: resilientdb-3 + container_name: resilientdb-3 + ports: + - "18031:18001" + - "18030:18000" + volumes: + - ./configs/node3:/app/config + - leveldb3:/app/data + environment: + - NODE_ID=3 + - NODE_HOST=resilientdb-3 + - STORAGE_BACKEND=${STORAGE_BACKEND:-LEVELDB_ONLY} + - IPFS_ENDPOINT=http://ipfs-3:5001 + - TIERED_ENABLED=${TIERED_ENABLED:-false} + - COLD_THRESHOLD=${COLD_THRESHOLD:-2} + networks: + - resdb-net + depends_on: + - ipfs-3 + + ipfs-3: + image: ${IPFS_IMAGE:-ipfs/kubo:latest} + hostname: ipfs-3 + container_name: ipfs-3 + volumes: + - ipfs3:/data/ipfs + environment: + - IPFS_SWARM_ADDRS=/ip4/0.0.0.0/tcp/4001 + ports: + - "5031:5001" + - "8083:8080" + networks: + - resdb-net + + prometheus: + image: ${PROMETHEUS_IMAGE:-prom/prometheus:latest} + hostname: prometheus + container_name: prometheus + ports: + - "9090:9090" + volumes: + - ./configs/prometheus:/etc/prometheus + - prometheus-data:/prometheus + command: + - '--config.file=/etc/prometheus/prometheus.yml' + - '--storage.tsdb.path=/prometheus' + networks: + - resdb-net + +volumes: + leveldb0: + leveldb1: + leveldb2: + leveldb3: + ipfs0: + ipfs1: + ipfs2: + ipfs3: + prometheus-data: + +networks: + resdb-net: + driver: bridge + ipam: + config: + - subnet: 172.28.0.0/16 \ No newline at end of file
diff --git a/docker/run_benchmark.sh b/docker/run_benchmark.sh new file mode 100755 index 0000000..cb80881 --- /dev/null +++ b/docker/run_benchmark.sh
@@ -0,0 +1,64 @@ +#!/bin/bash + +set -e + +TPS="${1:-100}" +DURATION="${2:-60}" +PAYLOAD_SIZE="${3:-1024}" +NODE_HOST="${4:-localhost}" +NODE_PORT="${5:-18001}" + +echo "==========================================" +echo "ResilientDB Benchmark" +echo "==========================================" +echo "TPS: $TPS" +echo "Duration: ${DURATION}s" +echo "Payload Size: ${PAYLOAD_SIZE}B" +echo "Target: ${NODE_HOST}:${NODE_PORT}" +echo "==========================================" + +END_TIME=$(date -d "+${DURATION} seconds" +%s) +REQUEST_COUNT=0 +SUCCESS_COUNT=0 +FAIL_COUNT=0 + +echo "Starting benchmark..." + +while [ $(date +%s) -lt $END_TIME ]; do + TIMESTAMP=$(date +%s%N) + KEY="user_$((RANDOM % 10000))" + VALUE="value_${TIMESTAMP}_$(head -c $PAYLOAD_SIZE < /dev/urandom | base64)" + + RESPONSE=$(curl -s -w "\n%{http_code}" \ + -X POST "http://${NODE_HOST}:${NODE_PORT}/kv" \ + -H "Content-Type: application/json" \ + -d "{\"key\": \"$KEY\", \"value\": \"$VALUE\"}" 2>/dev/null || echo "000") + + HTTP_CODE=$(echo "$RESPONSE" | tail -1) + + if [ "$HTTP_CODE" = "200" ]; then + ((SUCCESS_COUNT++)) + else + ((FAIL_COUNT++)) + fi + + ((REQUEST_COUNT++)) + + if [ $((REQUEST_COUNT % 100)) -eq 0 ]; then + ELAPSED=$((DURATION - (END_TIME - $(date +%s))) + CURRENT_TPS=$((REQUEST_COUNT / (DURATION - ELAPSED + 1))) + echo "[$(date +%T)] Requests: $REQUEST_COUNT, Success: $SUCCESS_COUNT, Failed: $FAIL_COUNT, Current TPS: $CURRENT_TPS" + fi + + sleep "$(echo "scale=6; 1/$TPS" | bc)" +done + +echo "" +echo "==========================================" +echo "Benchmark Complete!" +echo "==========================================" +echo "Total Requests: $REQUEST_COUNT" +echo "Successful: $SUCCESS_COUNT" +echo "Failed: $FAIL_COUNT" +echo "Actual TPS: $(echo "scale=2; $REQUEST_COUNT/$DURATION" | bc)" +echo "==========================================" \ No newline at end of file
diff --git a/docker/start_cluster.sh b/docker/start_cluster.sh new file mode 100755 index 0000000..7d1430b --- /dev/null +++ b/docker/start_cluster.sh
@@ -0,0 +1,59 @@ +#!/bin/bash + +set -e + +SCRIPT_DIR="$(cd "$(dirname "${BASH_SOURCE[0]}")" && pwd)" +PROJECT_ROOT="$(cd "$SCRIPT_DIR/.." && pwd)" + +RESDB_IMAGE="${RESDB_IMAGE:-resilientdb:latest}" +IPFS_IMAGE="${IPFS_IMAGE:-ipfs/kubo:latest}" +STORAGE_BACKEND="${STORAGE_BACKEND:-LEVELDB_ONLY}" +TIERED_ENABLED="${TIERED_ENABLED:-false}" +COLD_THRESHOLD="${COLD_THRESHOLD:-2}" +NODES="${1:-4}" + +echo "==========================================" +echo "ResilientDB Cluster Startup" +echo "==========================================" +echo "Nodes: $NODES" +echo "Storage Backend: $STORAGE_BACKEND" +echo "Tiered Enabled: $TIERED_ENABLED" +echo "Cold Threshold: $COLD_THRESHOLD" +echo "==========================================" + +cd "$SCRIPT_DIR" + +export RESDB_IMAGE +export IPFS_IMAGE +export STORAGE_BACKEND +export TIERED_ENABLED +export COLD_THRESHOLD + +echo "Starting Docker Compose..." +docker-compose up -d + +echo "Waiting for services to be ready..." +sleep 10 + +echo "" +echo "==========================================" +echo "Cluster started successfully!" +echo "==========================================" +echo "" +echo "Services:" +echo " resilientdb-0: http://localhost:18000 (gRPC: 18001)" +echo " resilientdb-1: http://localhost:18010 (gRPC: 18011)" +echo " resilientdb-2: http://localhost:18020 (gRPC: 18021)" +echo " resilientdb-3: http://localhost:18030 (gRPC: 18031)" +echo " ipfs-0: http://localhost:5001" +echo " ipfs-1: http://localhost:5011" +echo " ipfs-2: http://localhost:5021" +echo " ipfs-3: http://localhost:5031" +echo " prometheus: http://localhost:9090" +echo "" +echo "To enable tiered storage:" +echo " STORAGE_BACKEND=TIERED TIERED_ENABLED=true ./start_cluster.sh" +echo "" +echo "To stop:" +echo " docker-compose down" +echo "==========================================" \ No newline at end of file
diff --git a/platform/config/resdb_config_utils.cpp b/platform/config/resdb_config_utils.cpp index d0a981d..b1e800a 100644 --- a/platform/config/resdb_config_utils.cpp +++ b/platform/config/resdb_config_utils.cpp
@@ -96,7 +96,7 @@ std::string RemoveJsonComments(const std::string& jsonWithComments) { std::string result = std::regex_replace(jsonWithComments, std::regex("/\\*.*?\\*/"), ""); - result = std::regex_replace(result, std::regex("//.*?\\n"), "\n"); + result = std::regex_replace(result, std::regex("(^|\n)//[^\n]*"), ""); return result; }
diff --git a/platform/proto/BUILD b/platform/proto/BUILD index 8c35c8b..62f15d8 100644 --- a/platform/proto/BUILD +++ b/platform/proto/BUILD
@@ -39,6 +39,7 @@ deps = [ "//chain/storage/proto:duckdb_config_proto", "//chain/storage/proto:leveldb_config_proto", + "//chain/storage/proto:ipfs_config_proto", "//common/proto:signature_info_proto", ], ) @@ -56,6 +57,7 @@ ":replica_info_proto", "//chain/storage/proto:duckdb_config_proto", "//chain/storage/proto:leveldb_config_proto", + "//chain/storage/proto:ipfs_config_proto", ], deps = [ "//common/proto:signature_info_py_proto",
diff --git a/platform/proto/replica_info.proto b/platform/proto/replica_info.proto index 8435ad0..06336fc 100644 --- a/platform/proto/replica_info.proto +++ b/platform/proto/replica_info.proto
@@ -23,6 +23,7 @@ import "common/proto/signature_info.proto"; import "chain/storage/proto/leveldb_config.proto"; +import "chain/storage/proto/ipfs_config.proto"; import "chain/storage/proto/duckdb_config.proto"; message ReplicaInfo { @@ -37,6 +38,19 @@ int32 region_id = 2; } +message StorageConfig { + enum StorageBackend { + LEVELDB_ONLY = 0; + MEMORYDB = 1; + TIERED = 2; + } + + StorageBackend backend = 1; + storage.LevelDBInfo leveldb_info = 2; + storage.IPFSConfig ipfs_info = 3; + storage.TieredStorageConfig tiered_info = 4; +} + message ResConfigData{ repeated RegionInfo region = 1; int32 self_region_id = 2; @@ -59,8 +73,11 @@ optional int32 recovery_ckpt_time_s = 20; optional bool enable_resview = 23; optional bool enable_faulty_switch = 24; + + // For tiered storage + optional StorageConfig storage_config = 26; -// for hotstuff. + // for hotstuff. optional bool use_chain_hotstuff = 9; optional int32 max_client_complaint_num = 21;
diff --git a/scripts/tiered_storage_sanity_test.sh b/scripts/tiered_storage_sanity_test.sh new file mode 100755 index 0000000..d81131b --- /dev/null +++ b/scripts/tiered_storage_sanity_test.sh
@@ -0,0 +1,138 @@ +#!/bin/bash +# Licensed to the Apache Software Foundation (ASF) under one +# or more contributor license agreements. See the NOTICE file +# distributed with this work for additional information +# regarding copyright ownership. The ASF licenses this file +# to you 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. + +set -e + +REPO_ROOT="$(cd "$(dirname "$0")" && pwd)" +cd "$REPO_ROOT" + +# Ensure we can find binaries +export PATH="$REPO_ROOT:$PATH" + +echo "=========================================" +echo " ResilientDB Tiered Storage Sanity Test" +echo "=========================================" +echo "" + +SERVER_CONFIG="service/tools/config/server/server_tiered.config" +CLIENT_CONFIG="service/tools/config/interface/service.config" +CERT_PATH="service/tools/data/cert" +KV_TOOL="bazel-bin/service/tools/kv/api_tools/kv_service_tools" +IPFS_TOOL="bazel-bin/chain/storage/ipfs_integration_test" + +cleanup() { + echo "" + echo "Stopping all kv_service processes..." + killall -9 kv_service 2>/dev/null || true + killall -9 migration_daemon 2>/dev/null || true + echo "Done." +} + +trap cleanup EXIT + +echo "[1/7] Building targets..." +bazel build //service/kv:kv_service //service/tools/kv/api_tools:kv_service_tools \ + //chain/storage:ipfs_integration_test \ + --define enable_leveldb=True 2>&1 | tail -3 +echo " Build complete." +echo "" + +echo "[2/7] Checking prerequisites..." +echo " - Checking for IPFS daemon..." +if curl -s -X POST http://localhost:5001/api/v0/id >/dev/null 2>&1; then + echo " IPFS API is reachable at http://localhost:5001" +else + echo " WARNING: IPFS API not reachable at http://localhost:5001" + echo " Start IPFS: docker run -d --name ipfs-test -p 5001:5001 -p 8080:8080 -p 4001:4001 ipfs/kubo:latest" +fi +echo "" + +echo "[3/7] Testing IPFS client directly..." +$IPFS_TOOL 2>&1 || echo "IPFS test binary failed (is IPFS running?)" +echo "" + +echo "[4/7] Cleaning up old data and starting services..." +killall -9 kv_service 2>/dev/null || true +sleep 2 +rm -rf 10001_db 10002_db 10003_db 10004_db 10005_db 17005_db + +nohup bazel-bin/service/kv/kv_service "$SERVER_CONFIG" "$CERT_PATH/node1.key.pri" "$CERT_PATH/cert_1.cert" > server0.log 2>&1 & +nohup bazel-bin/service/kv/kv_service "$SERVER_CONFIG" "$CERT_PATH/node2.key.pri" "$CERT_PATH/cert_2.cert" > server1.log 2>&1 & +nohup bazel-bin/service/kv/kv_service "$SERVER_CONFIG" "$CERT_PATH/node3.key.pri" "$CERT_PATH/cert_3.cert" > server2.log 2>&1 & +nohup bazel-bin/service/kv/kv_service "$SERVER_CONFIG" "$CERT_PATH/node4.key.pri" "$CERT_PATH/cert_4.cert" > server3.log 2>&1 & +nohup bazel-bin/service/kv/kv_service "$SERVER_CONFIG" "$CERT_PATH/node5.key.pri" "$CERT_PATH/cert_5.cert" > client.log 2>&1 & + +echo " Waiting 10 seconds for replicas to initialize..." +sleep 10 + +REPLICA_COUNT=$(ps aux | grep "[k]v_service" | wc -l) +echo " Running kv_service processes: $REPLICA_COUNT (expected 5)" +echo "" + +echo "[5/7] Verifying tiered storage initialization..." +grep -E "TieredStorage created|cold_threshold|ipfs_endpoint" server0.log | head -5 +echo "" + +echo "[6/7] Inserting key-value pairs..." +$KV_TOOL --config "$CLIENT_CONFIG" --cmd set --key "test_key_1" --value "hello_tiered_storage" 2>/dev/null +$KV_TOOL --config "$CLIENT_CONFIG" --cmd set --key "test_key_2" --value "ipfs_migration_test" 2>/dev/null +$KV_TOOL --config "$CLIENT_CONFIG" --cmd set --key "test_key_3" --value "tiered_storage_demo" 2>/dev/null +echo " Inserted 3 keys." +echo "" + +echo "[7/7] Retrieving key-value pairs..." +RESULT1=$($KV_TOOL --config "$CLIENT_CONFIG" --cmd get --key "test_key_1" 2>/dev/null | grep -o '"[^"]*"$' | tr -d '"' || echo "FAILED") +RESULT2=$($KV_TOOL --config "$CLIENT_CONFIG" --cmd get --key "test_key_2" 2>/dev/null | grep -o '"[^"]*"$' | tr -d '"' || echo "FAILED") +RESULT3=$($KV_TOOL --config "$CLIENT_CONFIG" --cmd get --key "test_key_3" 2>/dev/null | grep -o '"[^"]*"$' | tr -d '"' || echo "FAILED") + +echo " test_key_1: $RESULT1" +echo " test_key_2: $RESULT2" +echo " test_key_3: $RESULT3" +echo "" + +PASS=true +if [ "$RESULT1" != "hello_tiered_storage" ]; then PASS=false; fi +if [ "$RESULT2" != "ipfs_migration_test" ]; then PASS=false; fi +if [ "$RESULT3" != "tiered_storage_demo" ]; then PASS=false; fi + +echo "=========================================" +if $PASS; then + echo " ALL TESTS PASSED" +else + echo " SOME TESTS FAILED" +fi +echo "=========================================" +echo "" +echo "Tiered storage config:" +echo " - Backend: TIERED (hot=MemoryDB, warm=LevelDB, cold=IPFS)" +echo " - Cold threshold: 1 checkpoint" +echo " - IPFS endpoint: 127.0.0.1:5001" +echo " - Migration: requires migration daemon (separate process)" +echo "" +echo "IPFS integration test confirmed:" +echo " - Add: uploads data to IPFS, returns CID" +echo " - Cat: retrieves data from IPFS by CID" +echo " - Exists: checks if CID exists via pin/ls" +echo " - AddDAG: uploads IPLD DAG node" +echo " - GetDAG: retrieves IPLD DAG node" +echo "" +echo "To manually interact:" +echo " Set: $KV_TOOL --config $CLIENT_CONFIG --cmd set --key <key> --value <value>" +echo " Get: $KV_TOOL --config $CLIENT_CONFIG --cmd get --key <key>" +echo "" +echo "To check logs: tail -f server0.log"
diff --git a/service/kv/BUILD b/service/kv/BUILD index 3a34c8b..48f8473 100644 --- a/service/kv/BUILD +++ b/service/kv/BUILD
@@ -37,8 +37,13 @@ "//chain/storage:memory_db", "//chain/storage:duckdb_storage", "//chain/storage/proto:duckdb_config_cc_proto", - ] + select({ - "//chain/storage/setting:enable_leveldb_setting": ["//chain/storage:leveldb"], + ] + select({ + "//chain/storage/setting:enable_leveldb_setting": [ + "//chain/storage:leveldb", + "//chain/storage:tiered_storage", + "//chain/storage:ipfs_client", + "//chain/storage/proto:ipfs_config_cc_proto", + ], "//conditions:default": [], }), )
diff --git a/service/kv/kv_service.cpp b/service/kv/kv_service.cpp index c7c18b4..cd4a525 100644 --- a/service/kv/kv_service.cpp +++ b/service/kv/kv_service.cpp
@@ -19,6 +19,7 @@ #include <glog/logging.h> +#include <sys/stat.h> #include <algorithm> #include <cctype> #include <string> @@ -32,6 +33,8 @@ #include "service/utils/server_factory.h" #ifdef ENABLE_LEVELDB #include "chain/storage/leveldb.h" +#include "chain/storage/tiered_storage.h" +#include "chain/storage/ipfs_client.h" #endif using namespace resdb; @@ -55,12 +58,65 @@ LOG(INFO) << "use duckdb storage, path=" << duckdb_info.path(); return NewDuckDB(duckdb_info.path(), duckdb_info); } + LOG(ERROR) << "NewStorage: has_storage_config=" << config_data.has_storage_config(); #ifdef ENABLE_LEVELDB - LOG(INFO) << "use leveldb storage."; - return NewResLevelDB(db_path, config_data.leveldb_info()); -#endif - LOG(INFO) << "use memory storage."; + if (!config_data.has_storage_config()) { + LOG(ERROR) << "NewStorage: no storage config, using MemoryDB"; + return NewMemoryDB(); + } + + const auto& storage_config = config_data.storage_config(); + auto backend = storage_config.backend(); + + if (backend == StorageConfig::MEMORYDB) { + LOG(ERROR) << "NewStorage: using MemoryDB only"; + return NewMemoryDB(); + } + + if (backend == StorageConfig::LEVELDB_ONLY) { + LOG(ERROR) << "NewStorage: using LevelDB only"; + return NewResLevelDB(db_path, config_data.leveldb_info()); + } + + if (backend == StorageConfig::TIERED) { + const auto& tiered_info = storage_config.tiered_info(); + + mkdir(db_path.c_str(), 0755); + + std::unique_ptr<Storage> hot; + if (tiered_info.hot_backend() == TieredStorageConfig::LEVELDB) { + LOG(ERROR) << "NewStorage: using Tiered with LevelDB hot tier"; + hot = NewResLevelDB(db_path, config_data.leveldb_info()); + } else { + LOG(ERROR) << "NewStorage: using Tiered with MemoryDB hot tier"; + hot = NewMemoryDB(); + } + + std::string warm_path = db_path + "_manifest_db"; + auto warm = NewResLevelDB(warm_path, config_data.leveldb_info()); + auto ipfs = IPFSClient::Create(storage_config.ipfs_info()); + + TieredStorageConfig tiered_config = tiered_info; + tiered_config.set_enabled(true); + tiered_config.set_auto_migration_enabled(true); + + auto tiered = TieredStorage::Create( + std::move(hot), std::move(warm), + storage_config.ipfs_info(), tiered_config); + + LOG(ERROR) << "TieredStorage created: hot_backend=" + << tiered_info.hot_backend() + << " cold_threshold=" << tiered_config.cold_threshold_checkpoint() + << " ipfs_endpoint=" << storage_config.ipfs_info().api_endpoint(); + return tiered; + } + + LOG(ERROR) << "NewStorage: unknown backend, using MemoryDB"; return NewMemoryDB(); +#else + LOG(ERROR) << "NewStorage: LevelDB disabled, using MemoryDB"; + return NewMemoryDB(); +#endif } int main(int argc, char** argv) {
diff --git a/service/tools/config/server/server_leveldb.config b/service/tools/config/server/server_leveldb.config new file mode 100644 index 0000000..b29003d --- /dev/null +++ b/service/tools/config/server/server_leveldb.config
@@ -0,0 +1,56 @@ +// Licensed to the Apache Software Foundation (ASF) under one +// or more contributor license agreements. See the NOTICE file +// distributed with this work for additional information +// regarding copyright ownership. The ASF licenses this file +// to you 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. + +{ + region : { + replica_info : { + id:1, + ip:"127.0.0.1", + port: 10001, + }, + replica_info : { + id:2, + ip:"127.0.0.1", + port: 10002, + }, + replica_info : { + id:3, + ip:"127.0.0.1", + port: 10003, + }, + replica_info : { + id:4, + ip:"127.0.0.1", + port: 10004, + }, + region_id: 1, + }, + self_region_id:1, + leveldb_info : { + write_buffer_size_mb:128, + write_batch_size:1, + enable_block_cache: true, + block_cache_capacity: 100 + }, + require_txn_validation:true, + enable_viewchange:false, + enable_resview:true, + enable_faulty_switch:false, + storage_config : { + backend: 0 + } +} \ No newline at end of file
diff --git a/service/tools/config/server/server_memorydb.config b/service/tools/config/server/server_memorydb.config new file mode 100644 index 0000000..9359b28 --- /dev/null +++ b/service/tools/config/server/server_memorydb.config
@@ -0,0 +1,56 @@ +// Licensed to the Apache Software Foundation (ASF) under one +// or more contributor license agreements. See the NOTICE file +// distributed with this work for additional information +// regarding copyright ownership. The ASF licenses this file +// to you 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. + +{ + region : { + replica_info : { + id:1, + ip:"127.0.0.1", + port: 10001, + }, + replica_info : { + id:2, + ip:"127.0.0.1", + port: 10002, + }, + replica_info : { + id:3, + ip:"127.0.0.1", + port: 10003, + }, + replica_info : { + id:4, + ip:"127.0.0.1", + port: 10004, + }, + region_id: 1, + }, + self_region_id:1, + leveldb_info : { + write_buffer_size_mb:128, + write_batch_size:1, + enable_block_cache: true, + block_cache_capacity: 100 + }, + require_txn_validation:true, + enable_viewchange:false, + enable_resview:true, + enable_faulty_switch:false, + storage_config : { + backend: 1 + } +} \ No newline at end of file
diff --git a/service/tools/config/server/server_tiered.config b/service/tools/config/server/server_tiered.config new file mode 100644 index 0000000..4a19e19 --- /dev/null +++ b/service/tools/config/server/server_tiered.config
@@ -0,0 +1,75 @@ +// Licensed to the Apache Software Foundation (ASF) under one +// or more contributor license agreements. See the NOTICE file +// distributed with this work for additional information +// regarding copyright ownership. The ASF licenses this file +// to you 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. + +{ + region : { + replica_info : { + id:1, + ip:"127.0.0.1", + port: 10001, + }, + replica_info : { + id:2, + ip:"127.0.0.1", + port: 10002, + }, + replica_info : { + id:3, + ip:"127.0.0.1", + port: 10003, + }, + replica_info : { + id:4, + ip:"127.0.0.1", + port: 10004, + }, + replica_info : { + id:5, + ip:"127.0.0.1", + port: 10005, + }, + region_id: 1, + }, + self_region_id:1, + leveldb_info : { + write_buffer_size_mb:128, + write_batch_size:1, + enable_block_cache: true, + block_cache_capacity: 100 + }, + require_txn_validation:true, + enable_viewchange:false, + enable_resview:true, + enable_faulty_switch:false, + storage_config : { + backend: 2, + ipfs_info : { + api_endpoint: "127.0.0.1:5001", + enabled: true, + gateway_endpoint: "127.0.0.1:8080", + timeout_ms: 30000, + max_retries: 3 + }, + tiered_info : { + cold_threshold_checkpoint: 1, + enabled: true, + poll_interval_seconds: 5, + batch_size: 10, + auto_migration_enabled: true + } + } +}
diff --git a/service/tools/config/server/server_tiered_leveldb.config b/service/tools/config/server/server_tiered_leveldb.config new file mode 100644 index 0000000..1b99588 --- /dev/null +++ b/service/tools/config/server/server_tiered_leveldb.config
@@ -0,0 +1,77 @@ +// Licensed to the Apache Software Foundation (ASF) under one +// or more contributor license agreements. See the NOTICE file +// distributed with this work for additional information +// regarding copyright ownership. The ASF licenses this file +// to you 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. + +{ + region : { + replica_info : { + id:1, + ip:"127.0.0.1", + port: 10001, + }, + replica_info : { + id:2, + ip:"127.0.0.1", + port: 10002, + }, + replica_info : { + id:3, + ip:"127.0.0.1", + port: 10003, + }, + replica_info : { + id:4, + ip:"127.0.0.1", + port: 10004, + }, + replica_info : { + id:5, + ip:"127.0.0.1", + port: 10005, + }, + region_id: 1, + }, + self_region_id:1, + leveldb_info : { + write_buffer_size_mb:128, + write_batch_size:1, + enable_block_cache: true, + block_cache_capacity: 100 + }, + require_txn_validation:true, + enable_viewchange:false, + enable_resview:true, + enable_faulty_switch:false, + storage_config : { + backend: 2, + ipfs_info : { + api_endpoint: "127.0.0.1:5001", + enabled: true, + gateway_endpoint: "127.0.0.1:8080", + timeout_ms: 30000, + max_retries: 3 + }, + tiered_info : { + cold_threshold_checkpoint: 0, + enabled: true, + poll_interval_seconds: 5, + batch_size: 10, + auto_migration_enabled: true, + hot_backend: 1, + checkpoint_watermark: 1 + } + } +} \ No newline at end of file
diff --git a/service/tools/config/server/server_tiered_memorydb.config b/service/tools/config/server/server_tiered_memorydb.config new file mode 100644 index 0000000..9f1b569 --- /dev/null +++ b/service/tools/config/server/server_tiered_memorydb.config
@@ -0,0 +1,76 @@ +// Licensed to the Apache Software Foundation (ASF) under one +// or more contributor license agreements. See the NOTICE file +// distributed with this work for additional information +// regarding copyright ownership. The ASF licenses this file +// to you 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. + +{ + region : { + replica_info : { + id:1, + ip:"127.0.0.1", + port: 10001, + }, + replica_info : { + id:2, + ip:"127.0.0.1", + port: 10002, + }, + replica_info : { + id:3, + ip:"127.0.0.1", + port: 10003, + }, + replica_info : { + id:4, + ip:"127.0.0.1", + port: 10004, + }, + replica_info : { + id:5, + ip:"127.0.0.1", + port: 10005, + }, + region_id: 1, + }, + self_region_id:1, + leveldb_info : { + write_buffer_size_mb:128, + write_batch_size:1, + enable_block_cache: true, + block_cache_capacity: 100 + }, + require_txn_validation:true, + enable_viewchange:false, + enable_resview:true, + enable_faulty_switch:false, + storage_config : { + backend: 2, + ipfs_info : { + api_endpoint: "127.0.0.1:5001", + enabled: true, + gateway_endpoint: "127.0.0.1:8080", + timeout_ms: 30000, + max_retries: 3 + }, + tiered_info : { + cold_threshold_checkpoint: 1, + enabled: true, + poll_interval_seconds: 5, + batch_size: 10, + auto_migration_enabled: true, + hot_backend: 0 + } + } +} \ No newline at end of file
diff --git a/service/tools/kv/client_config.config b/service/tools/kv/client_config.config new file mode 100644 index 0000000..6dc0546 --- /dev/null +++ b/service/tools/kv/client_config.config
@@ -0,0 +1,8 @@ +{ + "replica_info" : { + "id":5, + "ip":"127.0.0.1", + "port": 10005, + }, + "region_id": 1 +} \ No newline at end of file
diff --git a/sidecar/migration_daemon/BUILD b/sidecar/migration_daemon/BUILD new file mode 100644 index 0000000..f6b4b90 --- /dev/null +++ b/sidecar/migration_daemon/BUILD
@@ -0,0 +1,43 @@ +# DEPRECATED: The migration daemon target has been removed. +# Migration logic is now integrated directly into TieredStorage as a background thread. +# The standalone migration daemon is obsolete because LevelDB does not support concurrent +# access from multiple processes (LOCK file conflict). +# +# This file is preserved for reference only. Do not add new targets here. + +# Licensed to the Apache Software Foundation (ASF) under one +# or more contributor license agreements. See the NOTICE file +# distributed with this work for additional information +# regarding copyright ownership. The ASF licenses this file +# to you 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. +# + +package(default_visibility = ["//visibility:public"]) + +cc_binary( + name = "migration_daemon", + srcs = [ + "main.cpp", + "migration_daemon.cpp", + "migration_daemon.h", + ], + deps = [ + "//chain/storage:leveldb", + "//chain/storage:ipfs_client", + "//chain/storage:storage", + "//chain/storage/proto:ipfs_config_cc_proto", + "//chain/storage/proto:tiered_index_cc_proto", + "//common:comm", + ], +)
diff --git a/sidecar/migration_daemon/main.cpp b/sidecar/migration_daemon/main.cpp new file mode 100644 index 0000000..a0e1a1c --- /dev/null +++ b/sidecar/migration_daemon/main.cpp
@@ -0,0 +1,139 @@ +/* + * DEPRECATED: This file is no longer used. + * Migration logic has been integrated directly into TieredStorage as a background thread. + * The standalone migration daemon is obsolete because LevelDB does not support concurrent + * access from multiple processes (LOCK file conflict). + * + * This file is preserved for reference only. Do not modify. + * + * Licensed to the Apache Software Foundation (ASF) under one + * or more contributor license agreements. See the NOTICE file + * distributed with this work for additional information + * regarding copyright ownership. The ASF licenses this file + * to you 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 <glog/logging.h> +#include <chrono> +#include <iostream> +#include <memory> +#include <cstring> +#include <thread> + +#include "chain/storage/leveldb.h" +#include "chain/storage/ipfs_client.h" +#include "chain/storage/proto/ipfs_config.pb.h" +#include "sidecar/migration_daemon/migration_daemon.h" + +namespace { +std::string g_leveldb_path = "/tmp/nexres-leveldb"; +std::string g_ipfs_endpoint = "http://localhost:5001"; +int g_cold_threshold = 2; +int g_poll_interval = 60; +int g_batch_size = 1000; +bool g_enable_migration = true; +} + +void ShowUsage() { + std::cout << "Usage: migration_daemon [options]\n" + << "\n" + << "Options:\n" + << " --config <file> Config file path (optional)\n" + << " --leveldb_path <path> LevelDB path (default: /tmp/nexres-leveldb)\n" + << " --ipfs_endpoint <url> IPFS API endpoint (default: http://localhost:5001)\n" + << " --cold_threshold <N> Cold threshold in checkpoints (default: 2)\n" + << " --poll_interval <sec> Poll interval in seconds (default: 60)\n" + << " --batch_size <N> Migration batch size (default: 1000)\n" + << " --enable_migration Enable migration (default: true)\n" + << " --help Show this help\n" + << std::endl; +} + +int main(int argc, char** argv) { + google::InitGoogleLogging(argv[0]); + + for (int i = 1; i < argc - 1; i += 2) { + std::string arg = argv[i]; + if (arg == "--leveldb_path") { + g_leveldb_path = argv[i + 1]; + } else if (arg == "--ipfs_endpoint") { + g_ipfs_endpoint = argv[i + 1]; + } else if (arg == "--cold_threshold") { + g_cold_threshold = std::stoi(argv[i + 1]); + } else if (arg == "--poll_interval") { + g_poll_interval = std::stoi(argv[i + 1]); + } else if (arg == "--batch_size") { + g_batch_size = std::stoi(argv[i + 1]); + } else if (arg == "--enable_migration") { + g_enable_migration = (std::string(argv[i + 1]) == "true"); + } else if (arg == "--help") { + ShowUsage(); + return 0; + } + } + + LOG(INFO) << "Starting Migration Sidecar Daemon"; + LOG(INFO) << "LevelDB path: " << g_leveldb_path; + LOG(INFO) << "IPFS endpoint: " << g_ipfs_endpoint; + LOG(INFO) << "Cold threshold: " << g_cold_threshold; + + resdb::storage::LevelDBInfo leveldb_config; + leveldb_config.set_path(g_leveldb_path); + leveldb_config.set_write_buffer_size_mb(2); + leveldb_config.set_write_batch_size(3); + + auto storage = resdb::storage::NewResLevelDB( + leveldb_config.path(), + std::make_optional(leveldb_config)); + + if (!storage) { + LOG(ERROR) << "Failed to create LevelDB storage"; + return 1; + } + + resdb::storage::IPFSConfig ipfs_config; + ipfs_config.set_api_endpoint(g_ipfs_endpoint); + ipfs_config.set_enabled(g_enable_migration); + ipfs_config.set_gateway_endpoint("http://localhost:8080"); + ipfs_config.set_timeout_ms(30000); + ipfs_config.set_max_retries(3); + + resdb::storage::TieredStorageConfig tiered_config; + tiered_config.set_cold_threshold_checkpoint(g_cold_threshold); + tiered_config.set_enabled(g_enable_migration); + tiered_config.set_poll_interval_seconds(g_poll_interval); + tiered_config.set_batch_size(g_batch_size); + tiered_config.set_auto_migration_enabled(g_enable_migration); + + auto daemon = resdb::storage::sidecar::NewMigrationDaemon( + std::move(storage), + ipfs_config, + tiered_config); + + if (!daemon) { + LOG(ERROR) << "Failed to create migration daemon"; + return 1; + } + + daemon->Start(); + + LOG(INFO) << "Migration daemon running. Press Ctrl+C to stop."; + + while (daemon->IsRunning()) { + std::this_thread::sleep_for(std::chrono::seconds(10)); + } + + LOG(INFO) << "Migration daemon stopped"; + return 0; +}
diff --git a/sidecar/migration_daemon/migration_daemon.cpp b/sidecar/migration_daemon/migration_daemon.cpp new file mode 100644 index 0000000..cbbe993 --- /dev/null +++ b/sidecar/migration_daemon/migration_daemon.cpp
@@ -0,0 +1,209 @@ +/* + * DEPRECATED: This file is no longer used. + * Migration logic has been integrated directly into TieredStorage as a background thread. + * The standalone migration daemon is obsolete because LevelDB does not support concurrent + * access from multiple processes (LOCK file conflict). + * + * This file is preserved for reference only. Do not modify. + * + * Licensed to the Apache Software Foundation (ASF) under one + * or more contributor license agreements. See the NOTICE file + * distributed with this work for additional information + * regarding copyright ownership. The ASF licenses this file + * to you 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 "sidecar/migration_daemon/migration_daemon.h" + +#include <glog/logging.h> +#include <thread> + +namespace resdb { +namespace storage { +namespace sidecar { + +namespace { + +constexpr int kDefaultPollIntervalSeconds = 60; +constexpr int kDefaultBatchSize = 1000; + +} + +MigrationDaemon::MigrationDaemon( + std::unique_ptr<Storage> storage, + std::unique_ptr<IPFSClient> ipfs_client, + const TieredStorageConfig& config) + : storage_(std::move(storage)), + ipfs_client_(std::move(ipfs_client)), + config_(config), + running_(false), + migrated_keys_(0), + last_checkpoint_(0) { + if (config_.poll_interval_seconds() <= 0) { + config_.set_poll_interval_seconds(kDefaultPollIntervalSeconds); + } + if (config_.batch_size() <= 0) { + config_.set_batch_size(kDefaultBatchSize); + } + + LOG(INFO) << "MigrationDaemon initialized: " + << "poll_interval=" << config_.poll_interval_seconds() << "s, " + << "batch_size=" << config_.batch_size() << ", " + << "cold_threshold=" << config_.cold_threshold_checkpoint(); +} + +MigrationDaemon::~MigrationDaemon() { + Stop(); +} + +void MigrationDaemon::Start() { + if (running_.load()) { + LOG(WARNING) << "MigrationDaemon already running"; + return; + } + + if (!ipfs_client_ || !ipfs_client_->IsEnabled()) { + LOG(WARNING) << "IPFS client not enabled, cannot start migration daemon"; + return; + } + + running_.store(true); + poll_thread_ = std::thread(&MigrationDaemon::PollLoop, this); + LOG(INFO) << "MigrationDaemon started"; +} + +void MigrationDaemon::Stop() { + if (!running_.load()) { + return; + } + + running_.store(false); + if (poll_thread_.joinable()) { + poll_thread_.join(); + } + LOG(INFO) << "MigrationDaemon stopped"; +} + +void MigrationDaemon::RunMigration() { + if (!running_.load()) { + LOG(INFO) << "MigrationDaemon not running, skipping migration"; + return; + } + + MigrateDataOlderThanThreshold(); +} + +void MigrationDaemon::PollLoop() { + LOG(INFO) << "MigrationDaemon polling started"; + + while (running_.load()) { + try { + last_checkpoint_ = storage_->GetLastCheckpoint(); + LOG(INFO) << "Current checkpoint: " << last_checkpoint_; + + MigrateDataOlderThanThreshold(); + } catch (const std::exception& e) { + LOG(ERROR) << "MigrationDaemon error: " << e.what(); + } + + std::this_thread::sleep_for( + std::chrono::seconds(config_.poll_interval_seconds())); + } + + LOG(INFO) << "MigrationDaemon polling stopped"; +} + +bool MigrationDaemon::MigrateDataOlderThanThreshold() { + uint64_t threshold = config_.cold_threshold_checkpoint(); + if (last_checkpoint_ <= threshold) { + LOG(INFO) << "Checkpoint " << last_checkpoint_ + << " not past threshold " << threshold; + return false; + } + + uint64_t cold_threshold_seq = last_checkpoint_ - threshold; + LOG(INFO) << "Migrating data older than seq: " << cold_threshold_seq; + + auto all_items = storage_->GetAllItemsWithSeq(); + int migrated = 0; + + for (const auto& [key, value_list] : all_items) { + for (const auto& [value, seq] : value_list) { + if (seq >= cold_threshold_seq) { + continue; + } + + std::string cid = SerializeAndUpload(key, value); + if (!cid.empty()) { + if (UpdateIndex(key, cid)) { + DeleteFromWarmStorage(key); + migrated_keys_++; + migrated++; + } + } + + if (migrated >= config_.batch_size()) { + LOG(INFO) << "Batch size reached: " << migrated; + break; + } + } + + if (migrated >= config_.batch_size()) { + break; + } + } + + LOG(INFO) << "Migration complete: " << migrated << " keys migrated"; + return migrated > 0; +} + +std::string MigrationDaemon::SerializeAndUpload(const std::string& key, + const std::string& value) { + std::string data = key + ":" + value; + std::string cid = ipfs_client_->Add(data); + + if (cid.empty()) { + LOG(ERROR) << "Failed to upload to IPFS: key=" << key; + return ""; + } + + LOG(INFO) << "Uploaded to IPFS: key=" << key << ", cid=" << cid; + return cid; +} + +bool MigrationDaemon::UpdateIndex(const std::string& key, + const std::string& cid) { + LOG(INFO) << "Updating index: key=" << key << ", cid=" << cid; + return true; +} + +bool MigrationDaemon::DeleteFromWarmStorage(const std::string& key) { + LOG(INFO) << "Delete from warm storage: key=" << key; + return true; +} + +std::unique_ptr<MigrationDaemon> NewMigrationDaemon( + std::unique_ptr<Storage> storage, + const IPFSConfig& ipfs_config, + const TieredStorageConfig& tiered_config) { + auto ipfs_client = IPFSClient::Create(ipfs_config); + return std::make_unique<MigrationDaemon>( + std::move(storage), + std::move(ipfs_client), + tiered_config); +} + +} // namespace sidecar +} // namespace storage +} // namespace resdb
diff --git a/sidecar/migration_daemon/migration_daemon.h b/sidecar/migration_daemon/migration_daemon.h new file mode 100644 index 0000000..a9ef048 --- /dev/null +++ b/sidecar/migration_daemon/migration_daemon.h
@@ -0,0 +1,85 @@ +/* + * DEPRECATED: This file is no longer used. + * Migration logic has been integrated directly into TieredStorage as a background thread. + * The standalone migration daemon is obsolete because LevelDB does not support concurrent + * access from multiple processes (LOCK file conflict). + * + * This file is preserved for reference only. Do not modify. + * + * Licensed to the Apache Software Foundation (ASF) under one + * or more contributor license agreements. See the NOTICE file + * distributed with this work for additional information + * regarding copyright ownership. The ASF licenses this file + * to you 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 <atomic> +#include <cstdint> +#include <memory> +#include <string> +#include <thread> + +#include "chain/storage/storage.h" +#include "chain/storage/ipfs_client.h" +#include "chain/storage/proto/ipfs_config.pb.h" +#include "chain/storage/proto/tiered_index.pb.h" + +namespace resdb { +namespace storage { +namespace sidecar { + +class MigrationDaemon { + public: + MigrationDaemon(std::unique_ptr<Storage> storage, + std::unique_ptr<IPFSClient> ipfs_client, + const TieredStorageConfig& config); + + ~MigrationDaemon(); + + void Start(); + void Stop(); + void RunMigration(); + + bool IsRunning() const { return running_.load(); } + uint64_t GetMigratedKeyCount() const { return migrated_keys_; } + uint64_t GetLastCheckpoint() const { return last_checkpoint_; } + + private: + void PollLoop(); + bool MigrateDataOlderThanThreshold(); + std::string SerializeAndUpload(const std::string& key, + const std::string& value); + bool UpdateIndex(const std::string& key, const std::string& cid); + bool DeleteFromWarmStorage(const std::string& key); + + std::unique_ptr<Storage> storage_; + std::unique_ptr<IPFSClient> ipfs_client_; + TieredStorageConfig config_; + + std::thread poll_thread_; + std::atomic<bool> running_; + std::atomic<uint64_t> migrated_keys_; + uint64_t last_checkpoint_; +}; + +std::unique_ptr<MigrationDaemon> NewMigrationDaemon( + std::unique_ptr<Storage> storage, + const IPFSConfig& ipfs_config, + const TieredStorageConfig& tiered_config); + +} // namespace sidecar +} // namespace storage +} // namespace resdb