// Copyright (c) 2011 The LevelDB Authors. All rights reserved.

// Use of this source code is governed by a BSD-style license that can be

// found in the LICENSE file. See the AUTHORS file for names of contributors.

#include "db/log_reader.h"

#include <stdio.h>

#include "leveldb/env.h"

#include "util/coding.h"

#include "util/crc32c.h"

namespace leveldb {

namespace log {

Reader::Reporter::~Reporter() = default;

Reader::Reader(SequentialFile* file, Reporter* reporter, bool checksum,

               uint64_t initial_offset)

    : file_(file),

      reporter_(reporter),

      checksum_(checksum),

      backing_store_(new char[kBlockSize]),

      buffer_(),

      eof_(false),

      last_record_offset_(0),

      end_of_buffer_offset_(0),

      initial_offset_(initial_offset),

      resyncing_(initial_offset > 0) {}

Reader::~Reader() { delete[] backing_store_; }

bool Reader::SkipToInitialBlock() {

  const size_t offset_in_block = initial_offset_ % kBlockSize;

  uint64_t block_start_location = initial_offset_ - offset_in_block;

  // Don't search a block if we'd be in the trailer

  if (offset_in_block > kBlockSize - 6) {

    block_start_location += kBlockSize;

  }

  end_of_buffer_offset_ = block_start_location;

  // Skip to start of first block that can contain the initial record

  if (block_start_location > 0) {

    Status skip_status = file_->Skip(block_start_location);

    if (!skip_status.ok()) {

      ReportDrop(block_start_location, skip_status);

      return false;

    }

  }

  return true;

}

bool Reader::ReadRecord(Slice* record, std::string* scratch) {

  if (last_record_offset_ < initial_offset_) {

    if (!SkipToInitialBlock()) {

      return false;

    }

  }

  scratch->clear();

  record->clear();

  bool in_fragmented_record = false;

  // Record offset of the logical record that we're reading

  // 0 is a dummy value to make compilers happy

  uint64_t prospective_record_offset = 0;

  Slice fragment;

  while (true) {

    const unsigned int record_type = ReadPhysicalRecord(&fragment);

    // ReadPhysicalRecord may have only had an empty trailer remaining in its

    // internal buffer. Calculate the offset of the next physical record now

    // that it has returned, properly accounting for its header size.

    uint64_t physical_record_offset =

        end_of_buffer_offset_ - buffer_.size() - kHeaderSize - fragment.size();

    if (resyncing_) {

      if (record_type == kMiddleType) {

        continue;

      } else if (record_type == kLastType) {

        resyncing_ = false;

        continue;

      } else {

        resyncing_ = false;

      }

    }

    switch (record_type) {

      case kFullType:

        if (in_fragmented_record) {

          // Handle bug in earlier versions of log::Writer where

          // it could emit an empty kFirstType record at the tail end

          // of a block followed by a kFullType or kFirstType record

          // at the beginning of the next block.

          if (!scratch->empty()) {

            ReportCorruption(scratch->size(), "partial record without end(1)");

          }

        }

        prospective_record_offset = physical_record_offset;

        scratch->clear();

        *record = fragment;

        last_record_offset_ = prospective_record_offset;

        return true;

      case kFirstType:

        if (in_fragmented_record) {

          // Handle bug in earlier versions of log::Writer where

          // it could emit an empty kFirstType record at the tail end

          // of a block followed by a kFullType or kFirstType record

          // at the beginning of the next block.

          if (!scratch->empty()) {

            ReportCorruption(scratch->size(), "partial record without end(2)");

          }

        }

        prospective_record_offset = physical_record_offset;

        scratch->assign(fragment.data(), fragment.size());

        in_fragmented_record = true;

        break;

      case kMiddleType:

        if (!in_fragmented_record) {

          ReportCorruption(fragment.size(),

                           "missing start of fragmented record(1)");

        } else {

          scratch->append(fragment.data(), fragment.size());

        }

        break;

      case kLastType:

        if (!in_fragmented_record) {

          ReportCorruption(fragment.size(),

                           "missing start of fragmented record(2)");

        } else {

          scratch->append(fragment.data(), fragment.size());

          *record = Slice(*scratch);

          last_record_offset_ = prospective_record_offset;

          return true;

        }

        break;

      case kEof:

        if (in_fragmented_record) {

          // This can be caused by the writer dying immediately after

          // writing a physical record but before completing the next; don't

          // treat it as a corruption, just ignore the entire logical record.

          scratch->clear();

        }

        return false;

      case kBadRecord:

        if (in_fragmented_record) {

          ReportCorruption(scratch->size(), "error in middle of record");

          in_fragmented_record = false;

          scratch->clear();

        }

        break;

      default: {

        char buf[40];

        snprintf(buf, sizeof(buf), "unknown record type %u", record_type);

        ReportCorruption(

            (fragment.size() + (in_fragmented_record ? scratch->size() : 0)),

            buf);

        in_fragmented_record = false;

        scratch->clear();

        break;

      }

    }

  }

  return false;

}

uint64_t Reader::LastRecordOffset() { return last_record_offset_; }

void Reader::ReportCorruption(uint64_t bytes, const char* reason) {

  ReportDrop(bytes, Status::Corruption(reason, file_->GetName()));

}

void Reader::ReportDrop(uint64_t bytes, const Status& reason) {

  if (reporter_ != nullptr &&

      end_of_buffer_offset_ - buffer_.size() - bytes >= initial_offset_) {

    reporter_->Corruption(static_cast<size_t>(bytes), reason);

  }

}

unsigned int Reader::ReadPhysicalRecord(Slice* result) {

  while (true) {

    if (buffer_.size() < kHeaderSize) {

      if (!eof_) {

        // Last read was a full read, so this is a trailer to skip

        buffer_.clear();

        Status status = file_->Read(kBlockSize, &buffer_, backing_store_);

        end_of_buffer_offset_ += buffer_.size();

        if (!status.ok()) {

          buffer_.clear();

          ReportDrop(kBlockSize, status);

          eof_ = true;

          return kEof;

        } else if (buffer_.size() < kBlockSize) {

          eof_ = true;

        }

        continue;

      } else {

        // Note that if buffer_ is non-empty, we have a truncated header at the

        // end of the file, which can be caused by the writer crashing in the

        // middle of writing the header. Instead of considering this an error,

        // just report EOF.

        buffer_.clear();

        return kEof;

      }

    }

    // Parse the header

    const char* header = buffer_.data();

    const uint32_t a = static_cast<uint32_t>(header[4]) & 0xff;

    const uint32_t b = static_cast<uint32_t>(header[5]) & 0xff;

    const unsigned int type = header[6];

    const uint32_t length = a | (b << 8);

    if (kHeaderSize + length > buffer_.size()) {

      size_t drop_size = buffer_.size();

      buffer_.clear();

      if (!eof_) {

        ReportCorruption(drop_size, "bad record length");

        return kBadRecord;

      }

      // If the end of the file has been reached without reading |length| bytes

      // of payload, assume the writer died in the middle of writing the record.

      // Don't report a corruption.

      return kEof;

    }

    if (type == kZeroType && length == 0) {

      // Skip zero length record without reporting any drops since

      // such records are produced by the mmap based writing code in

      // env_posix.cc that preallocates file regions.

      buffer_.clear();

      return kBadRecord;

    }

    // Check crc

    if (checksum_) {

      uint32_t expected_crc = crc32c::Unmask(DecodeFixed32(header));

      uint32_t actual_crc = crc32c::Value(header + 6, 1 + length);

      if (actual_crc != expected_crc) {

        // Drop the rest of the buffer since "length" itself may have

        // been corrupted and if we trust it, we could find some

        // fragment of a real log record that just happens to look

        // like a valid log record.

        size_t drop_size = buffer_.size();

        buffer_.clear();

        ReportCorruption(drop_size, "checksum mismatch");

        return kBadRecord;

      }

    }

    buffer_.remove_prefix(kHeaderSize + length);

    // Skip physical record that started before initial_offset_

    if (end_of_buffer_offset_ - buffer_.size() - kHeaderSize - length <

        initial_offset_) {

      result->clear();

      return kBadRecord;

    }

    *result = Slice(header + kHeaderSize, length);

    return type;

  }

}

}  // namespace log

}  // namespace leveldb