storage.h

Go to the documentation of this file.

#pragma once
 
#include "librats/util/rats_export.h"
#include "librats/node/peer_network.h"
#include "librats/peer/peer.h"
#include "librats/peer/peer_id.h"
#include "librats/core/bytes.h"
#include "librats/util/json.h"
 
#include <string>
#include <vector>
#include <functional>
#include <map>
#include <memory>
#include <mutex>
#include <unordered_map>
#include <atomic>
#include <chrono>
#include <thread>
#include <optional>
#include <condition_variable>
 
namespace librats {
 
enum class StorageValueType : uint8_t {
    BINARY = 0x01,      // Raw binary data
    STRING = 0x02,      // UTF-8 string
    INT64 = 0x03,       // 64-bit signed integer
    DOUBLE = 0x04,      // 64-bit floating point
    JSON = 0x05         // JSON document
};
 
enum class StorageOperation : uint8_t {
    OP_PUT = 0x01,         // Insert or update
    OP_DELETE = 0x02       // Delete key
};
 
enum class StorageSyncStatus {
    NOT_STARTED,        // Sync not initiated
    IN_PROGRESS,        // Sync currently running
    COMPLETED,          // Sync completed successfully
    FAILED              // Sync failed
};
 
struct RATS_API StorageEntry {
    std::string key;                    // Key string
    StorageValueType type;              // Value type
    std::vector<uint8_t> data;          // Serialized value data
    uint64_t timestamp_ms;              // Unix timestamp in milliseconds (for LWW)
    std::string origin_peer_id;         // Peer that created/modified this entry (hex PeerId)
    uint32_t checksum;                  // CRC32 checksum for integrity
    bool deleted;                       // Tombstone marker for deleted entries
 
    StorageEntry()
        : type(StorageValueType::BINARY),
          timestamp_ms(0),
          checksum(0),
          deleted(false) {}
 
    StorageEntry(const std::string& k, StorageValueType t,
                 const std::vector<uint8_t>& d, uint64_t ts,
                 const std::string& peer_id)
        : key(k), type(t), data(d), timestamp_ms(ts),
          origin_peer_id(peer_id), checksum(0), deleted(false) {
        calculate_checksum();
    }
 
    // Calculate CRC32 checksum
    void calculate_checksum();
 
    // Verify checksum
    bool verify_checksum() const;
 
    // Serialize entry to binary format
    std::vector<uint8_t> serialize() const;
 
    void serialize_into(std::vector<uint8_t>& out) const;
 
    size_t serialized_size() const;
 
    static bool deserialize(const uint8_t* data, size_t size, size_t offset,
                            StorageEntry& entry, size_t& bytes_read);
 
    static bool deserialize(const std::vector<uint8_t>& data, size_t offset,
                           StorageEntry& entry, size_t& bytes_read);
 
    // Compare for LWW resolution (returns true if this entry wins)
    bool wins_over(const StorageEntry& other) const;
};
 
struct StorageChangeEvent {
    StorageOperation operation;         // PUT or DELETE
    std::string key;                    // Affected key
    StorageValueType type;              // Value type (for PUT)
    std::vector<uint8_t> old_data;      // Previous value (if any)
    std::vector<uint8_t> new_data;      // New value (for PUT)
    uint64_t timestamp_ms;              // Operation timestamp
    std::string origin_peer_id;         // Peer that made the change
    bool is_remote;                     // True if change came from another peer
};
 
struct StorageConfig {
    static constexpr uint32_t kMaxValueSize      = 2 * 1024 * 1024;
    static constexpr uint32_t kMaxSyncBatchBytes = 2 * 1024 * 1024;
    static constexpr uint32_t kMinSyncBatchBytes = 16 * 1024;
 
    std::string data_directory;         // Directory for storage files
    std::string database_name;          // Database filename prefix
    bool enable_sync;                   // Enable network synchronization
    uint32_t compaction_threshold;      // Number of tombstones before compaction
    uint32_t max_value_size;            // Maximum value size in bytes (<= kMaxValueSize)
    bool persist_to_disk;               // Whether to persist data to disk
    uint32_t sync_batch_bytes;
    uint32_t sync_min_interval_ms;
 
    StorageConfig()
        : data_directory("./storage"),
          database_name("rats_storage"),
          enable_sync(true),
          compaction_threshold(1000),
          max_value_size(1024 * 1024),        // 1 MiB max value size
          persist_to_disk(true),
          sync_batch_bytes(256 * 1024),       // 256 KiB per snapshot chunk
          sync_min_interval_ms(5000) {}
};
 
struct StorageStatistics {
    size_t total_entries;               // Total number of entries
    size_t deleted_entries;             // Number of tombstones
    uint64_t total_data_bytes;          // Total size of stored data
    uint64_t disk_usage_bytes;          // Disk space used
    uint64_t entries_synced;            // Entries synced from peers
    uint64_t entries_sent;              // Entries sent to peers
    uint64_t sync_requests_received;    // Number of sync requests received
    uint64_t sync_requests_sent;        // Number of sync requests sent
    uint64_t sync_chunks_sent;          // Snapshot chunks put on the wire
    uint64_t sync_chunks_received;      // Snapshot chunks applied from peers
    uint64_t resyncs_scheduled;         // Snapshots owed to peers that filled up
    std::chrono::steady_clock::time_point last_sync_time;  // Last sync timestamp
    StorageSyncStatus sync_status;      // Current sync status
};
 
using StorageChangeCallback = std::function<void(const StorageChangeEvent&)>;
using StorageSyncCompleteCallback = std::function<void(bool success, const std::string& error_message)>;
 
class RATS_API StorageManager final : public Subsystem {
public:
    explicit StorageManager(const StorageConfig& config = StorageConfig());
 
    ~StorageManager() override;
 
    StorageManager(const StorageManager&) = delete;
    StorageManager& operator=(const StorageManager&) = delete;
 
    // =========================================================================
    // Subsystem
    // =========================================================================
 
    void attach(NodeContext& ctx) override;
    void start() override;
    void stop() override;
 
    // =========================================================================
    // Configuration
    // =========================================================================
 
    void set_config(const StorageConfig& config);
    const StorageConfig& get_config() const;
 
    // =========================================================================
    // Put Operations (Write)
    // =========================================================================
 
    bool put(const std::string& key, const std::string& value);
    bool put(const std::string& key, int64_t value);
    bool put(const std::string& key, double value);
    bool put(const std::string& key, const std::vector<uint8_t>& value);
    bool put_json(const std::string& key, const librats::Json& value);
 
    // =========================================================================
    // Get Operations (Read)
    // =========================================================================
 
    std::optional<std::string> get_string(const std::string& key) const;
    std::optional<int64_t> get_int(const std::string& key) const;
    std::optional<double> get_double(const std::string& key) const;
    std::optional<std::vector<uint8_t>> get_binary(const std::string& key) const;
    std::optional<librats::Json> get_json(const std::string& key) const;
    std::optional<StorageValueType> get_type(const std::string& key) const;
 
    // =========================================================================
    // Delete and Query Operations
    // =========================================================================
 
    bool remove(const std::string& key);
    bool has(const std::string& key) const;
    std::vector<std::string> keys() const;
    std::vector<std::string> keys_with_prefix(const std::string& prefix) const;
    size_t size() const;
    bool empty() const;
    void clear();
 
    // =========================================================================
    // Persistence Operations
    // =========================================================================
 
    bool save();
    bool load();
    size_t compact();
 
    // =========================================================================
    // Synchronization Operations
    // =========================================================================
 
    bool request_sync();
    StorageSyncStatus get_sync_status() const;
    bool is_synced() const;
 
    // =========================================================================
    // Event Callbacks
    // =========================================================================
 
    void set_change_callback(StorageChangeCallback callback);
    void set_sync_complete_callback(StorageSyncCompleteCallback callback);
 
    // =========================================================================
    // Statistics
    // =========================================================================
 
    StorageStatistics get_statistics() const;
    librats::Json get_statistics_json() const;
 
private:
    // Network message handlers (run on a reactor thread).
    void on_storage_message(const PeerId& from, ByteView payload);
    void on_peer_connected(const PeerId& peer_id);
    void on_peer_disconnected(const PeerId& peer_id);
    void on_peer_writable(const PeerId& peer_id);
 
    PeerNetwork* network_ = nullptr;
    StorageConfig config_;
 
    // In-memory storage.
    //
    // Ordered, not hashed, and that is load-bearing: a snapshot is streamed one
    // bounded chunk at a time, and between chunks the only thing carried over is
    // the last key sent. An ordered map turns that key back into a position with
    // upper_bound(), so a stream costs nothing to keep alive and cannot be
    // derailed by concurrent writes — an unordered_map would need either a
    // materialised key list per peer or iterators that a rehash invalidates.
    // It also makes keys_with_prefix() a range scan instead of a full walk.
    mutable std::mutex storage_mutex_;
    std::map<std::string, StorageEntry> entries_;
 
    // Sync state.
    //
    // Lock order, exactly one rule: storage_mutex_ and sync_mutex_ are never held
    // at the same time, and neither is ever held across a call into PeerNetwork.
    mutable std::mutex sync_mutex_;
    StorageSyncStatus sync_status_;
    bool initial_sync_complete_;
    std::chrono::steady_clock::time_point last_sync_time_;
 
    struct PeerSync {
        bool                                  streaming = false;  
        bool                                  started   = false;  
        std::string                           cursor;             
        bool                                  owed      = false;  
        std::chrono::steady_clock::time_point last_start{};       
    };
    std::unordered_map<PeerId, PeerSync, PeerId::Hash> peers_;    
    uint64_t sync_epoch_ = 0;
 
    // The sync thread: serializes and paces every snapshot chunk, so no reactor
    // thread ever does work proportional to the size of the database.
    std::atomic<bool>       sync_running_{false};
    std::thread             sync_thread_;
    std::condition_variable sync_cv_;   
    // Sync tuning, copied out of the config when the thread starts and read-only
    // afterwards, so the thread never races a set_config() on another thread.
    size_t                    batch_bytes_{0};
    std::chrono::milliseconds sync_interval_{0};
 
    // Statistics
    mutable std::mutex stats_mutex_;
    StorageStatistics stats_;
 
    // Callbacks
    StorageChangeCallback change_callback_;
    StorageSyncCompleteCallback sync_complete_callback_;
 
    // Background persistence thread
    std::atomic<bool> running_;
    std::thread persistence_thread_;
    std::condition_variable persistence_cv_;
    std::mutex persistence_mutex_;
    bool dirty_;  // Flag indicating unsaved changes
 
    // Wire opcodes (MessageType::Storage payload, byte 0)
    static constexpr uint8_t OP_ENTRY        = 1;
    static constexpr uint8_t OP_SYNC_REQUEST = 2;
    static constexpr uint8_t OP_SYNC_CHUNK   = 3;
 
    // SYNC_CHUNK flags (byte 1)
    static constexpr uint8_t FLAG_LAST = 0x01;   
 
    static constexpr size_t kMaxInboundMessage = 8 * 1024 * 1024;
 
    // Private methods
    void initialize();
    void shutdown();
    void persistence_thread_loop();
    void start_sync_thread();
    void stop_sync_thread();
    void sync_thread_loop();
    enum class ChunkResult {
        Continue,   
        Blocked,    
        Finished    
    };
    ChunkResult stream_snapshot_chunk(const PeerId& peer);
    void schedule_snapshot(const PeerId& peer, bool requested_by_peer);
 
    // Internal put with full control
    bool put_internal(const std::string& key, StorageValueType type,
                     const std::vector<uint8_t>& data,
                     uint64_t timestamp_ms = 0,
                     const std::string& origin_peer_id = "",
                     bool broadcast = true);
 
    // Serialization helpers
    std::vector<uint8_t> serialize_value(int64_t value) const;
    std::vector<uint8_t> serialize_value(double value) const;
    std::vector<uint8_t> serialize_value(const std::string& value) const;
    int64_t deserialize_int64(const std::vector<uint8_t>& data) const;
    double deserialize_double(const std::vector<uint8_t>& data) const;
    std::string deserialize_string(const std::vector<uint8_t>& data) const;
 
    // Network operations. Every one of them honours send()'s return value: a peer
    // that answers "no room" is owed a snapshot instead of further entries.
    void replicate_entry(const StorageEntry& entry, const PeerId* except);
    void broadcast_entry(const StorageEntry& entry) { replicate_entry(entry, nullptr); }
    void forward_entry(const StorageEntry& entry, const PeerId& except) {
        replicate_entry(entry, &except);
    }
    void send_sync_request(const PeerId& peer_id);
 
    // Apply a remote entry with LWW; fills `out_event` and returns true if applied.
    bool apply_remote_entry(const StorageEntry& entry, StorageChangeEvent* out_event);
    uint32_t apply_chunk(const PeerId& from, const uint8_t* data, size_t size, uint32_t count);
 
    // File path helpers
    std::string get_data_file_path() const;
    std::string get_index_file_path() const;
 
    // Disk I/O
    bool write_data_file();
    bool read_data_file();
 
    // Utility
    static void sanitize_config(StorageConfig& config);
    uint64_t get_current_timestamp_ms() const;
    std::string get_our_peer_id() const;
    void notify_change(const StorageChangeEvent& event);
    void mark_dirty();
};
 
// CRC32 calculation utility function
uint32_t storage_calculate_crc32(const void* data, size_t length);
 
// Convert StorageValueType to string
std::string storage_value_type_to_string(StorageValueType type);
 
// Convert string to StorageValueType
StorageValueType string_to_storage_value_type(const std::string& str);
 
} // namespace librats