src/core/record.cpp - master - smychkov/SStorage

src/core/record.cpp
168 строк7 KB
Андрей Смычков
docs: перевод оставшихся комментариев на русский + актуализация AGENTS
25 апр 2026, 10:08
25 апр 2026, 10:08f2b9fd9
#include "record.hpp"
#include "../util/varint.hpp"

namespace sstorage {

//============================================================================
// Конструкторы
//============================================================================
Record::Record(std::string key, std::string value, uint64_t seqNo)
    : key_(std::move(key)), value_(std::move(value)),
      seqNo_(seqNo), tombstone_(false) {
}

Record Record::makeTombstone(std::string key, uint64_t seqNo) {
    Record r;
    r.key_ = std::move(key);
    r.seqNo_ = seqNo;
    r.tombstone_ = true;
    // value_ остаётся пустым
    return r;
}

//============================================================================
// Сравнение
//============================================================================
// LSM-порядок: ключи по возрастанию, при равных ключах — новые версии первые
// (seqNo DESC). Это позволяет при итерации взять первую запись для key и
// гарантированно получить самую свежую версию.
//============================================================================
bool Record::operator<(const Record& other) const {
    if (key_ != other.key_) {
        return key_ < other.key_;
    }
    // seqNo DESC внутри одного key
    return seqNo_ > other.seqNo_;
}

bool Record::operator==(const Record& other) const {
    return key_ == other.key_ && seqNo_ == other.seqNo_ &&
           tombstone_ == other.tombstone_ && value_ == other.value_;
}

//============================================================================
// Сериализация
//============================================================================
// Формат см. в record.hpp. Упаковка компактная — varint для всех длин и seqNo.
//============================================================================
void Record::serialize(std::string& out) const {
    // Ключ
    util::encodeVarint(key_.size(), out);
    out.append(key_);

    // Флаги (1 байт)
    uint8_t flags = tombstone_ ? static_cast<uint8_t>(RecordFlags::kTombstone) : 0;
    out.push_back(static_cast<char>(flags));

    // Значение (пропускается если tombstone)
    if (!tombstone_) {
        util::encodeVarint(value_.size(), out);
        out.append(value_);
    } else {
        // Для tombstone value опускается целиком — экономия байт
        // Десериализатор определит по флагу, что value отсутствует
    }

    // Монотонный номер записи (для разрешения конфликтов)
    util::encodeVarint(seqNo_, out);
}

//============================================================================
// Десериализация
//============================================================================
std::optional<Record> Record::deserialize(const char* data, size_t len,
                                          size_t& consumed) {
    consumed = 0;
    if (data == nullptr && len > 0) {
        return std::nullopt;
    }
    size_t offset = 0;

    // Лимит на размер ключа/значения внутри Record — защита от
    // целенаправленно сконструированных varint с очень большими значениями.
    // Это же значение используется в Database как максимум, поэтому любые
    // данные больше — заведомо битые или вредоносные.
    constexpr uint64_t kMaxFieldSize = 64ULL * 1024 * 1024;   // 64 МБ

    // 1. keyLen
    uint64_t keyLen;
    size_t n;
    if (!util::decodeVarint(data + offset, len - offset, keyLen, n)) {
        return std::nullopt;
    }
    offset += n;
    // Защита от integer underflow: offset может равняться len (валидный случай
    // "len - offset == 0"), но не превышать его. decodeVarint это гарантирует,
    // однако проверяем защитно.
    if (offset > len) return std::nullopt;
    // Защита от невалидно большого keyLen (больше оставшегося буфера
    // или выше разумного лимита).
    if (keyLen > kMaxFieldSize) return std::nullopt;
    if (keyLen > len - offset) return std::nullopt;

    Record r;
    r.key_.assign(data + offset, static_cast<size_t>(keyLen));
    offset += static_cast<size_t>(keyLen);

    // 2. flags
    if (offset >= len) {
        return std::nullopt;
    }
    uint8_t flags = static_cast<uint8_t>(data[offset]);
    ++offset;
    r.tombstone_ = (flags & static_cast<uint8_t>(RecordFlags::kTombstone)) != 0;

    // 3. value (только если не tombstone)
    if (!r.tombstone_) {
        uint64_t valueLen;
        if (!util::decodeVarint(data + offset, len - offset, valueLen, n)) {
            return std::nullopt;
        }
        offset += n;
        if (offset > len) return std::nullopt;
        if (valueLen > kMaxFieldSize) return std::nullopt;
        if (valueLen > len - offset) return std::nullopt;

        r.value_.assign(data + offset, static_cast<size_t>(valueLen));
        offset += static_cast<size_t>(valueLen);
    }

    // 4. seqNo
    if (offset > len) return std::nullopt;
    if (!util::decodeVarint(data + offset, len - offset, r.seqNo_, n)) {
        return std::nullopt;
    }
    offset += n;
    if (offset > len) return std::nullopt;

    consumed = offset;
    return r;
}

//============================================================================
// Размер в памяти
//============================================================================
// Учитываем:
//   - capacity двух std::string (обычно >= size)
//   - sizeof самого объекта (поля)
//   - 16 байт overhead для каждой std::string на heap (примерно)
//============================================================================
size_t Record::approximateSize() const {
    return sizeof(*this) + key_.size() + value_.size() + 32;
}

//============================================================================
// Размер в сериализованном виде
//============================================================================
size_t Record::serializedSize() const {
    size_t size = 0;
    size += util::varintLength(key_.size()) + key_.size();
    size += 1; // flags
    if (!tombstone_) {
        size += util::varintLength(value_.size()) + value_.size();
    }
    size += util::varintLength(seqNo_);
    return size;
}

}