logfile.hh

/**
 * Copyright (c) 2007-2012, Timothy Stack
 *
 * All rights reserved.
 *
 * Redistribution and use in source and binary forms, with or without
 * modification, are permitted provided that the following conditions are met:
 *
 * * Redistributions of source code must retain the above copyright notice, this
 * list of conditions and the following disclaimer.
 * * Redistributions in binary form must reproduce the above copyright notice,
 * this list of conditions and the following disclaimer in the documentation
 * and/or other materials provided with the distribution.
 * * Neither the name of Timothy Stack nor the names of its contributors
 * may be used to endorse or promote products derived from this software
 * without specific prior written permission.
 *
 * THIS SOFTWARE IS PROVIDED BY THE REGENTS AND CONTRIBUTORS ''AS IS'' AND ANY
 * EXPRESS OR IMPLIED WARRANTIES, INCLUDING, BUT NOT LIMITED TO, THE IMPLIED
 * WARRANTIES OF MERCHANTABILITY AND FITNESS FOR A PARTICULAR PURPOSE ARE
 * DISCLAIMED. IN NO EVENT SHALL THE REGENTS OR CONTRIBUTORS BE LIABLE FOR ANY
 * DIRECT, INDIRECT, INCIDENTAL, SPECIAL, EXEMPLARY, OR CONSEQUENTIAL DAMAGES
 * (INCLUDING, BUT NOT LIMITED TO, PROCUREMENT OF SUBSTITUTE GOODS OR SERVICES;
 * LOSS OF USE, DATA, OR PROFITS; OR BUSINESS INTERRUPTION) HOWEVER CAUSED AND
 * ON ANY THEORY OF LIABILITY, WHETHER IN CONTRACT, STRICT LIABILITY, OR TORT
 * (INCLUDING NEGLIGENCE OR OTHERWISE) ARISING IN ANY WAY OUT OF THE USE OF THIS
 * SOFTWARE, EVEN IF ADVISED OF THE POSSIBILITY OF SUCH DAMAGE.
 *
 * @file logfile.hh
 */

#ifndef logfile_hh
#define logfile_hh

#include <algorithm>
#include <atomic>
#include <chrono>
#include <deque>
#include <filesystem>
#include <optional>
#include <string>
#include <unordered_map>
#include <utility>
#include <vector>

#include <stdint.h>
#include <stdio.h>
#include <sys/stat.h>
#include <sys/types.h>

#include "ArenaAlloc/arenaalloc.h"
#include "base/auto_fd.hh"
#include "base/auto_mem.hh"
#include "base/lnav_log.hh"
#include "base/map_util.hh"
#include "base/progress.hh"
#include "base/result.h"
#include "bookmarks.hh"
#include "file_options.hh"
#include "line_buffer.hh"
#include "log_format_fwd.hh"
#include "logfile_fwd.hh"
#include "mapbox/variant.hpp"
#include "safe/safe.h"
#include "shared_buffer.hh"
#include "type_safe/strong_typedef.hpp"
#include "unique_path.hh"

/**
 * How far rebuild_index() has gotten with a file.
 *
 * Every field is atomic because the UI thread reads them while a worker is
 * writing them.  That is also why the numbers live here rather than being
 * derived from lf_index on demand: reading lf_index.size() while a worker is
 * appending to it races against a vector that may be reallocating.  A tick's
 * reading of these is handed to the UI as a vector of index_progress_report
 * (logfile_fwd.hh), which is where the drawing code gets its numbers from.
 */
struct index_progress {
    std::atomic<file_off_t> ip_offset{0};
    std::atomic<file_ssize_t> ip_total{0};
    std::atomic<bool> ip_done{false};
    /** Raised by the UI thread when the user interrupts the pass. */
    std::atomic<bool> ip_abort{false};
};

namespace lnav {

/**
 * @return How many workers to index `file_count` files across, from
 * /tuning/logfile/indexing-threads.  0 asks the machine for a count and 1
 * means index them one at a time.
 *
 * Also warms the injected singletons the scan path reaches through
 * function-local statics, while the UI thread is still the only one running.
 * Call it on the UI thread ahead of every fan-out.
 */
size_t logfile_indexing_width(size_t file_count);

/**
 * Decompress the given files into the line-buffer cache, several at a time.
 *
 * @param on_progress Called on this thread with the compressed bytes consumed
 *   so far and the total, or (0, 0) when none of the files can say.
 *   Returning interrupt stops every file that is still being cached.
 * @return False if on_progress interrupted the work.
 */
bool logfile_enable_caches(
    const std::vector<::logfile*>& files,
    const std::function<progress_result_t(file_off_t, file_off_t)>& on_progress
    = {});

/**
 * @return True while logfile_enable_caches() has workers filling the files'
 *   line buffers.  Its on_progress callback runs during that time, and must not
 *   draw anything that reads those files, like the views or breadcrumbs.
 */
bool logfile_caches_in_flight();

}  // namespace lnav

struct logfile_activity {
    // Stats for the indexing pipeline — wall/cpu time accumulate
    // across each rebuild_index() call that consumed bytes (empty
    // polls don't); memory_bytes is refreshed to a current snapshot.
    // The wall/cpu gap is roughly time spent blocked on the
    // underlying file (raw read, decompressor, remote fetch, etc.).
    struct index_stats {
        std::chrono::microseconds is_wall_us{};
        std::chrono::microseconds is_cpu_us{};
        size_t is_memory_bytes{0};
    };

    int64_t la_polls{0};
    int64_t la_reads{0};
    index_stats la_index;
    // Approximate per-file memory held by the line_buffer (I/O
    // buffer + per-line offset / utf / ansi / col-width vectors).
    // Tracked separately from index memory because the line_buffer
    // is an I/O cache with very different sizing dynamics than the
    // indexed view.
    size_t la_line_buffer_memory_bytes{0};
};

/**
 * Container for the lines in a log file and some metadata.
 */
class logfile
    : public unique_path_source
    , public std::enable_shared_from_this<logfile> {
public:
    using iterator = std::vector<logline>::iterator;
    using const_iterator = std::vector<logline>::const_iterator;

    struct metadata {
        text_format_t m_format;
        std::string m_value;
    };

    /**
     * The relationship between a descriptor passed to open() and the path.
     */
    enum class fd_source {
        /** The descriptor is not associated with the path, like stdin. */
        detached,
        /** The descriptor was opened from the path. */
        of_path,
    };

    /**
     * Construct a logfile with the given arguments.
     *
     * @param filename The name of the log file.
     * @param fd The file descriptor for accessing the file or -1 if the
     * constructor should open the file specified by 'filename'.  The
     * descriptor needs to be seekable.
     * @param src Whether 'fd' was opened from 'filename'.
     */
    static Result<std::shared_ptr<logfile>, std::string> open(
        std::filesystem::path filename,
        const logfile_open_options& loo,
        auto_fd fd = auto_fd{},
        fd_source src = fd_source::detached);

    logfile(const logfile&) = delete;
    logfile& operator=(const logfile&) = delete;

    ~logfile() override;

    const logfile_activity& get_activity() const { return this->lf_activity; }

    std::optional<std::filesystem::path> get_actual_path() const
    {
        return this->lf_actual_path;
    }

    /** @return The filename as given in the constructor. */
    const std::filesystem::path& get_filename() const
    {
        return this->lf_filename;
    }

    const std::string get_filename_as_string() const
    {
        return this->lf_filename_as_string;
    }

    std::filesystem::path get_path_for_key() const;

    /**
     * @return The key to use when looking for this file in
     *   file_collection::fc_name_to_stubs.  This is the path that the scan
     *   was working with, which is not necessarily the name the file is
     *   displayed under.
     */
    std::string get_stub_key() const;

    /** @return The filename as given in the constructor, excluding the path
     * prefix. */
    const std::string& get_basename() const { return this->lf_basename; }

    int get_fd() const { return this->lf_line_buffer.get_fd(); }

    /** @param filename The new filename for this log file. */
    void set_filename(const std::string& filename);

    static uint64_t next_serial()
    {
        static std::atomic<uint64_t> counter{0};

        return counter.fetch_add(1, std::memory_order_relaxed) + 1;
    }

    const std::string& get_content_id() const { return this->lf_content_id; }

    /**
     * @return An id for this file, handed out in creation order and never
     * reused.
     *
     * For code that needs to say "the same file" without holding the file
     * alive.  The address will not do: a new logfile can be allocated where a
     * freed one used to live, and would then compare equal to it.
     */
    uint64_t get_serial() const { return this->lf_serial; }

    /** @return The inode for this log file. */
    const struct stat& get_stat() const { return this->lf_stat; }

    time_t get_origin_mtime() const;

    size_t get_longest_line_length() const { return this->lf_longest_line; }

    bool is_compressed() const { return this->lf_line_buffer.is_compressed(); }

    bool has_line_metadata() const
    {
        return this->lf_line_buffer.has_line_metadata();
    }

    bool is_valid_filename() const { return this->lf_valid_filename; }

    file_off_t get_index_size() const { return this->lf_index_size; }

    /**
     * @return The amount of data in the (possibly compressed) file that has
     * been indexed.
     */
    file_off_t get_indexed_file_offset() const
    {
        return this->lf_line_buffer.get_read_offset(this->lf_index_size);
    }

    int get_index_generation() const { return this->lf_index_generation; }

    /** A line number in this file. */
    struct file_line
        : type_safe::strong_typedef<file_line, uint32_t>
        , type_safe::strong_typedef_op::equality_comparison<file_line>
        , type_safe::strong_typedef_op::relational_comparison<file_line>
        , type_safe::strong_typedef_op::subtraction<file_line> {
        using strong_typedef::strong_typedef;
    };

    /** [line], which is below MAX_LINES, so it fits. */
    static file_line to_file_line(size_t line)
    {
        return file_line(static_cast<uint32_t>(line));
    }

    struct file_mtime {};
    struct embedded_file_time {};

    using anchor_time_t
        = mapbox::util::variant<file_line, file_mtime, embedded_file_time>;

    anchor_time_t get_anchor_time() const { return this->lf_anchor_time; }

    /**
     * The log statement each line's message came from, and the stack frames
     * in it, kept by lnav::stmt_stats.  Cleared along with the index.
     */
    struct stmt_ids {
        /** Statement ids by line number; 0 for lines not looked at yet. */
        std::vector<uint32_t> si_ids;
        /**
         * Lines before this have been matched since the statements last
         * changed.  The rest keep what they had (if anything) until they're
         * matched again, so the counts stay available while a re-scan's
         * changes are applied.
         */
        size_t si_verified{0};

        /**
         * A stack frame's id: its index in lnav::stmt_stats' frame table,
         * which is also its log_stack_frames rowid.
         */
        struct frame_id
            : type_safe::strong_typedef<frame_id, uint32_t>
            , type_safe::strong_typedef_op::equality_comparison<frame_id>
            , type_safe::strong_typedef_op::relational_comparison<frame_id> {
            using strong_typedef::strong_typedef;
        };

        /** An offset into si_frame_ids. */
        struct frame_ids_offset
            : type_safe::strong_typedef<frame_ids_offset, uint32_t>
            , type_safe::strong_typedef_op::subtraction<frame_ids_offset> {
            using strong_typedef::strong_typedef;
        };

        /** The frames of one line, the end of its run in si_frame_ids. */
        struct frame_line {
            file_line fl_line;
            frame_ids_offset fl_end;
        };
        static_assert(sizeof(frame_line) == 8);

        /**
         * The ids of the stack frames in each line's message, from
         * lnav::stmt_stats' frame table.  Only the lines with frames are in
         * si_frame_lines, in order, each with where its run of
         * si_frame_ids ends; the run starts where the one before it ends.
         * Lines before si_frames_verified have been looked at.  A message's
         * frames don't depend on the statements, so a source re-scan leaves
         * them.
         */
        std::vector<frame_line> si_frame_lines;
        std::vector<frame_id> si_frame_ids;
        size_t si_frames_verified{0};

        /**
         * Where a mapping pass's time goes, for finding out why one is
         * slow, kept by lnav::stmt_stats for each file and for the whole
         * pass.  Each worker counts in its own copy, which is added to its
         * file's and the pass's once it's done with the file for the slice,
         * so the workers don't share these as they go.  The times are summed
         * over the workers, so they can add up to more than the pass took.
         * Logged and reset at the end of each pass, with log2src's own
         * stats, which break down the time in find_statement().  Not reset
         * by clear(), so a pass's counts include the lines mapped before the
         * index was thrown away.
         */
        struct counters {
            /** Lines looked at, and the ones skipped early. */
            uint64_t mc_lines{0};
            uint64_t mc_continued{0};
            uint64_t mc_no_format{0};
            /** Lines only looked at for their frames, after a re-scan. */
            uint64_t mc_frames_only{0};
            uint64_t mc_no_body{0};

            /** read_full_message() and erase_ansi(). */
            uint64_t mc_read_ns{0};
            uint64_t mc_read_bytes{0};
            uint64_t mc_max_message_bytes{0};
            uint64_t mc_annotate_ns{0};
            uint64_t mc_frames_ns{0};

            /**
             * The calls to log2src from match_line(), including the trip
             * across the FFI and making the JSON of the result; compare
             * with log2src's own time to see how much that costs.
             */
            uint64_t mc_find_calls{0};
            uint64_t mc_find_ns{0};
            uint64_t mc_max_find_ns{0};
            /** Hashing the statement's schema and getting its id. */
            uint64_t mc_intern_ns{0};
            /** Interns that weren't in the worker's own cache. */
            uint64_t mc_shared_interns{0};
            /** sample_vars(), and how many of its calls matched again. */
            uint64_t mc_sample_ns{0};
            uint64_t mc_samples{0};

            /** What the messages were mapped to. */
            uint64_t mc_statements{0};
            uint64_t mc_src_locations{0};
            uint64_t mc_messages{0};
            /** The data_parser for the messages without a statement. */
            uint64_t mc_parse_ns{0};

            /** Time in the workers, which is the time per file per slice. */
            uint64_t mc_worker_ns{0};

            void merge(const counters& rhs);
        };

        counters si_counters;

        void clear();

        /** Forgets every line's frames, so they are found again. */
        void reset_frames();

        /** Sets the frames of line si_frames_verified and moves past it. */
        void add_frames(const std::vector<frame_id>& ids);

        /** The range of si_frame_ids with the frames of [line]. */
        std::pair<size_t, size_t> frames_of(size_t line) const;

        /** Forgets the lines from [size] on, so they are matched again. */
        void truncate(size_t size);

        /** Drops the first [count] lines, as the index does. */
        void discard_front(size_t count);

    private:
        /** The first of si_frame_lines at or after [line]. */
        std::vector<frame_line>::const_iterator first_frame_line(
            file_line line) const;

        /** Where the run of [iter]'s line starts in si_frame_ids. */
        frame_ids_offset run_start(
            std::vector<frame_line>::const_iterator iter) const;
    };

    stmt_ids& get_stmt_ids() { return this->lf_stmt_ids; }

    const stmt_ids& get_stmt_ids() const { return this->lf_stmt_ids; }

    /**
     * One of the values the format's scan settled on for each line (its
     * thread id or its opid, scan_batch_context::sbc_line_ids_set), so the
     * log tables can answer log_thread_id and log_opid without reading and
     * parsing the message again.  The scan works both out anyway, for the
     * thread and opid time ranges.
     *
     * Kept as runs of lines with the same value: each run lasts until the
     * next one starts, and lines from ir_end on aren't known.  A run's id is
     * the value's index in ir_values plus one, or 0 when the scan didn't
     * record the lines (lines read while the format was being detected,
     * formats that don't fill it in), which means: read the message.  Kept
     * in step with lf_index like lf_stmt_ids.  Written only while the file
     * is being indexed.
     *
     * The values point into lf_allocator, where the scan keeps its copies of
     * thread ids and opids, so this is cleared whenever the arena is reset.
     */
    struct line_id_runs {
        struct run {
            uint32_t r_start;
            uint32_t r_id;
        };

        std::vector<run> ir_runs;
        uint32_t ir_end{0};
        /** The values by id minus one; [0] is invalid, for "no value". */
        std::vector<string_key> ir_values{string_key::invalid()};
        std::unordered_map<string_key, uint32_t, frag_hasher, frag_equal>
            ir_ids;
        /** The run of the last get(), since lookups mostly go in order. */
        mutable size_t ir_hint{0};

        void clear()
        {
            this->ir_runs.clear();
            this->ir_end = 0;
            this->ir_values.resize(1);
            this->ir_ids.clear();
            this->ir_hint = 0;
        }

        /** Forgets the lines from [size] on. */
        void truncate(size_t size);

        /**
         * The id for a value (nullopt for none).  The key's bytes must be in
         * lf_allocator.
         */
        uint32_t id_for(const std::optional<string_key>& key);

        /**
         * Records lines [from, to) as having the value with the id (0 for
         * not recorded).  Lines between the last recorded one and [from]
         * are not recorded.
         */
        void record(size_t from, size_t to, uint32_t id);

        /**
         * Records lines [from, to), continuations of the message before
         * them, as having its value.
         */
        void continue_to(size_t from, size_t to);

        /**
         * The line's value, an invalid key when it has none, or nullptr
         * when it wasn't recorded.
         */
        const string_key* get(size_t line) const;

    private:
        void push_run(uint32_t start, uint32_t id);
    };

    struct line_ids {
        line_id_runs li_tids;
        line_id_runs li_opids;

        void clear()
        {
            this->li_tids.clear();
            this->li_opids.clear();
        }

        void truncate(size_t size)
        {
            this->li_tids.truncate(size);
            this->li_opids.truncate(size);
        }

        /** Records lines [from, to) as having the thread id and opid. */
        void record(size_t from,
                    size_t to,
                    const std::optional<string_key>& tid,
                    const std::optional<string_key>& opid)
        {
            this->li_tids.record(from, to, this->li_tids.id_for(tid));
            this->li_opids.record(from, to, this->li_opids.id_for(opid));
        }

        /** Records lines [from, to) as not recorded. */
        void record_unknown(size_t from, size_t to)
        {
            this->li_tids.record(from, to, 0);
            this->li_opids.record(from, to, 0);
        }

        void continue_to(size_t from, size_t to)
        {
            this->li_tids.continue_to(from, to);
            this->li_opids.continue_to(from, to);
        }
    };

    const line_ids& get_line_ids() const { return this->lf_line_ids; }

    file_ssize_t get_content_size() const
    {
        auto lb_size = this->lf_line_buffer.get_file_size();
        if (lb_size != -1) {
            return lb_size;
        }
        return this->lf_stat.st_size;
    }

    /**
     * @return How far indexing has gotten, as (done, total), or nullopt when
     * the total cannot be established.
     *
     * The numerator is always lf_index_size -- decompressed bytes indexed --
     * because that is the thing that actually advances smoothly.  Only the
     * denominator is in question, and it is taken from, in order:
     *
     *   - the line buffer, once it knows the real size.  That covers plain
     *     files, pipes that have hit EOF, and any compressed file that has
     *     been read to the end.
     *   - a gzip trailer's ISIZE, which is exact below 4GB and, unlike the
     *     decompressor's own position, is known before any of the file has
     *     been read.  Discarded the moment indexing passes it, which is what
     *     catches a wrapped value or a concatenated file.
     *   - the on-disk size, for an uncompressed file whose size the line
     *     buffer has not settled yet.
     *
     * Anything left over -- bzip2, or a gzip whose trailer proved wrong --
     * has no honest denominator and reports nothing.
     */
    std::optional<std::pair<file_off_t, file_ssize_t>> get_index_progress()
        const;

    /**
     * @return Whether the whole file has been read and indexed.
     *
     * Deliberately not a comparison of the two numbers above:
     * is_data_available() already knows that a compressed file's size is
     * only settled once decompression reaches EOF.
     */
    bool is_fully_indexed() const
    {
        return !this->lf_indexing
            || (this->lf_activity.la_polls > 0
                && !this->lf_line_buffer.is_data_available(
                    this->lf_index_size, this->lf_stat.st_size));
    }

    std::optional<const_iterator> line_for_offset(file_off_t off) const;

    /**
     * @return The detected format, rebuild_index() must be called before this
     * will return a value other than NULL.
     */
    std::shared_ptr<log_format> get_format() const { return this->lf_format; }

    log_format* get_format_ptr() const { return this->lf_format.get(); }

    intern_string_t get_format_name() const;

    std::optional<text_format_t> get_text_format() const
    {
        return this->lf_text_format;
    }

    void set_text_format(std::optional<text_format_t> tf)
    {
        this->lf_text_format = tf;
    }

    std::chrono::microseconds get_modified_time() const
    {
        return this->lf_index_time;
    }

    int get_time_offset_line() const { return this->lf_time_offset_line; }

    const timeval& get_time_offset() const { return this->lf_time_offset; }

    void adjust_content_time(int line,
                             const struct timeval& tv,
                             bool abs_offset = true);

    void clear_time_offset()
    {
        timeval tv = {0, 0};

        this->adjust_content_time(-1, tv);
    }

    bool mark_as_duplicate(const std::string& name);

    const logfile_open_options& get_open_options() const
    {
        return this->lf_options;
    }

    void set_include_in_session(bool enabled)
    {
        this->lf_options.with_include_in_session(enabled);
    }

    void set_init_location(file_location_t loc)
    {
        this->lf_options.with_init_location(loc);
    }

    void reset_state();

    bool is_time_adjusted() const
    {
        return (this->lf_time_offset.tv_sec != 0
                || this->lf_time_offset.tv_usec != 0);
    }

    iterator begin() { return this->lf_index.begin(); }

    const_iterator begin() const { return this->lf_index.begin(); }

    const_iterator cbegin() const { return this->lf_index.begin(); }

    iterator end() { return this->lf_index.end(); }

    const_iterator end() const { return this->lf_index.end(); }

    const_iterator cend() const { return this->lf_index.end(); }

    /** @return The number of lines in the index. */
    size_t size() const { return this->lf_index.size(); }

    const_iterator find_from_time(std::chrono::microseconds us) const;

    /**
     * @return The indexes of the lines in time order, or an empty vector
     *   when the index is already in time order.
     */
    const std::vector<uint32_t>& get_time_order() const
    {
        return this->lf_time_order;
    }

    /**
     * @return The line with the lowest time from the given line to the end
     *   of the index.  The line must be less than size().
     */
    const logline& earliest_line_from(size_t line) const;

    /**
     * @return The position in the time order of the first line whose time
     *   is not less than the given time.
     */
    size_t time_order_lower_bound(std::chrono::microseconds us) const;

    logline& operator[](int index) { return this->lf_index[index]; }

    std::optional<const_iterator> find_line(int line_number) const
    {
        if (line_number < 0 || line_number >= this->lf_index.size()) {
            return std::nullopt;
        }
        return this->lf_index.begin() + line_number;
    }

    logline& at(int index) { return this->lf_index.at(index); }

    logline& front() { return this->lf_index.front(); }

    logline& back() { return this->lf_index.back(); }

    bool in_range() const;

    /** @return True if this log file still exists. */
    bool exists() const;

    void close() { this->lf_is_closed = true; }

    bool is_closed() const { return this->lf_is_closed; }

    timeval original_line_time(iterator ll);

    Result<shared_buffer_ref, std::string> read_line(iterator ll,
                                                     subline_options opts = {});

    enum class read_format_t {
        plain,
        with_framing,
    };

    struct read_file_result {
        file_range rfr_range;
        std::string rfr_content;
    };

    Result<read_file_result, std::string> read_file(read_format_t format);

    Result<shared_buffer_ref, std::string> read_range(const file_range& fr);

    /** The first of the lines a message was rendered into. */
    iterator line_base(iterator ll) { return ll - this->sub_line_index(ll); }

    const_iterator line_base(const_iterator ll) const
    {
        return ll - this->sub_line_index(ll);
    }

    /**
     * Which of its message's rendered lines this is, 0 for the first.  A far
     * sub-line is at least logline::SUB_LINE_STRIDE lines into its message,
     * so the lines are skipped that many at a time from one.  Lines are
     * discarded from the front of a streamed file without regard to
     * messages, so a jump never goes before the first line.
     */
    size_t sub_line_index(const_iterator ll) const
    {
        constexpr auto stride = logline::SUB_LINE_STRIDE;
        size_t retval = 0;

        while (ll->is_far_sub_line()
               && static_cast<size_t>(ll - this->cbegin()) >= stride)
        {
            ll -= stride;
            retval += stride;
        }
        while (ll != this->cbegin() && ll->is_sub_line()) {
            --ll;
            retval += 1;
        }

        return retval;
    }

    iterator message_start(iterator ll)
    {
        auto retval = this->line_base(ll);

        // An earlier message's rendered lines aren't messages either, so
        // they're passed over in one go.
        while (retval != this->begin() && !retval->is_message()) {
            retval = this->line_base(retval - 1);
        }

        return retval;
    }

    std::pair<iterator, iterator> message_lines(iterator ll);

    /**
     * @param ll The first line of a message.
     * @return The number of lines in the message, counting the first line
     *   and the continuation lines after it.
     */
    size_t message_line_count(const_iterator ll) const;

    struct message_length_result {
        file_ssize_t mlr_length;
        size_t mlr_line_count;
        file_range::metadata mlr_metadata;
    };

    message_length_result message_byte_length(const_iterator ll,
                                              bool include_continues = true);

    file_range get_file_range(const_iterator ll, bool include_continues = true)
    {
        auto mlr = this->message_byte_length(ll, include_continues);

        return {
            ll->get_offset(),
            mlr.mlr_length,
            mlr.mlr_metadata,
        };
    }

    file_range get_msg_range(const_iterator ll)
    {
        return this->get_file_range(ll, true);
    }

    file_off_t get_line_content_offset(const_iterator ll)
    {
        return ll->get_offset() + (this->lf_line_buffer.is_piper() ? 22 : 0);
    }

    size_t get_line_number(const_iterator ll) const;

    void read_full_message(const_iterator ll,
                           shared_buffer_ref& msg_out,
                           line_buffer::scan_direction dir
                           = line_buffer::scan_direction::forward,
                           read_format_t format = read_format_t::plain);

    Result<shared_buffer_ref, std::string> read_raw_message(const_iterator ll);

    enum class rebuild_result_t {
        INVALID,
        NO_NEW_LINES,
        NEW_LINES,
        NEW_ORDER,
    };

    /**
     * Index any new data in the log file.
     *
     * @param lo The observer object that will be called regularly during
     * indexing.
     * @return True if any new lines were indexed.
     */
    rebuild_result_t rebuild_index(std::optional<ui_clock::time_point> deadline
                                   = std::nullopt);

    void reobserve_from(iterator iter);

    /**
     * Zero the progress and seed the denominator, on the thread that is about
     * to hand this file to a worker.
     *
     * The total is seeded here rather than left to the worker so that the
     * aggregate the UI shows is complete from the first tick and only moves
     * forward; filled in as files were picked up, the bar would slide
     * backwards.  It comes from get_index_progress() so the numerator and
     * denominator stay in one coordinate space -- get_content_size() reports
     * a compressed file's on-disk size until the stream hits EOF and its
     * decompressed size after, which made the total jump mid-pass.
     */
    void begin_indexing_progress();

    /**
     * The reobserve_from() counterpart of begin_indexing_progress(): seeds
     * the same slots, in the same byte units, for a re-scan starting at
     * `iter`.
     */
    void begin_reobserve_progress(iterator iter);

    /** @see index_progress -- safe to read while a worker is scanning. */
    const index_progress& indexing_progress() const
    {
        return this->lf_index_progress;
    }

    /** Ask the scan of this file to stop as soon as it notices. */
    void abort_indexing()
    {
        this->lf_index_progress.ip_abort.store(true, std::memory_order_relaxed);
    }

    /** Called by the scanning thread when it is done with this file. */
    void finish_indexing_progress()
    {
        this->lf_index_progress.ip_done.store(true, std::memory_order_relaxed);
    }

    void set_logline_observer(logline_observer* llo);

    logline_observer* get_logline_observer() const
    {
        return this->lf_logline_observer;
    }

    bool operator<(const logfile& rhs) const
    {
        bool retval;

        if (this->lf_index.empty()) {
            retval = true;
        } else if (rhs.lf_index.empty()) {
            retval = false;
        } else {
            retval = this->lf_index[0] < rhs.lf_index[0];
        }

        return retval;
    }

    bool is_indexing() const { return this->lf_indexing; }

    void set_indexing(bool val) { this->lf_indexing = val; }

    /** Check the invariants for this object. */
    bool invariant()
    {
        require(!this->lf_filename.empty());

        return true;
    }

    std::filesystem::path get_path() const override;

    /**
     * The most lines that will be indexed for a single file.  Kept equal to
     * lnav::logfile::MAX_LINES in logfile.cfg.hh, which the configuration
     * uses, without this header having to include that one.
     */
    static constexpr uint64_t MAX_LINES = 1ULL << 27;

    /**
     * While fewer lines than this have been indexed, a better matching format
     * can still replace the current one, which reindexes the file from the
     * start.
     */
    static constexpr size_t RETRY_MATCH_SIZE = 250;

    /**
     * Formats that describe themselves with a header need to read this many
     * lines before they can match, so candidates are not skipped based on
     * their file type until the index has gone past this point.
     */
    static constexpr size_t FILE_TYPE_PRUNE_SIZE = 20;

    enum class note_type {
        indexing_disabled,
        duplicate,
        not_utf,
        line_limit,
    };

    using note_map = lnav::map::small<note_type, lnav::console::user_message>;
    using safe_notes = safe::Safe<note_map, std::recursive_mutex>;

    note_map get_notes() const { return *this->lf_notes.readAccess(); }

    const std::vector<logline_value_stats>& get_value_stats() const
    {
        return this->lf_value_stats;
    }

    const logline_value_stats* stats_for_value(intern_string_t name) const;

    log_format_file_state get_format_file_state() const
    {
        return {
            this->lf_value_stats,
            this->lf_pattern_locks,
            this->lf_time_scanner,
        };
    }

    /**
     * @return The scanner that parsed this file's timestamps.  Anything
     * rendering or re-parsing one of its lines wants this rather than the
     * format's, which only holds the configuration it was seeded from.
     */
    date_time_scanner& get_time_scanner() const
    {
        return this->lf_time_scanner;
    }

    using safe_opid_state = safe::Safe<log_opid_state, std::recursive_mutex>;

    safe_opid_state& get_opids() { return this->lf_opids; }

    using safe_thread_id_state = safe::Safe<log_thread_id_state>;

    safe_thread_id_state& get_thread_ids() { return this->lf_thread_ids; }

    void set_logline_opid(uint32_t line_number, string_fragment opid);

    void set_opid_description(string_fragment opid, string_fragment desc);

    /**
     * Bumped when a file's opid map is changed outside of indexing (a line's
     * opid or an opid's description set by the user), which the LOG view's
     * index doesn't record.  The all_opids table keys its cache on it.
     */
    static std::atomic<uint64_t> ID_EDITS;

    void clear_logline_opid(uint32_t line_number);

    void quiesce() { this->lf_line_buffer.quiesce(); }

    bool needs_cache() const { return this->lf_line_buffer.needs_cache(); }

    void enable_cache(const line_buffer::cache_progress_cb& on_progress = {})
    {
        this->lf_line_buffer.enable_cache(on_progress);
    }

    void dump_stats();

    robin_hood::unordered_map<uint32_t, bookmark_metadata>&
    get_bookmark_metadata()
    {
        return this->lf_bookmark_metadata;
    }

    /**
     * False if no line has been given a user's opid, so the bookmark
     * metadata doesn't need to be checked for one.  It is not reset when
     * the opids are cleared.
     */
    bool may_have_user_opids() const { return this->lf_may_have_user_opids; }

    std::map<std::string, metadata>& get_embedded_metadata()
    {
        return this->lf_embedded_metadata;
    }

    const std::map<std::string, metadata>& get_embedded_metadata() const
    {
        return this->lf_embedded_metadata;
    }

    std::optional<std::pair<std::string, lnav::file_options>> get_file_options()
        const
    {
        return this->lf_file_options;
    }

    const robin_hood::unordered_set<intern_string_t, intern_hasher>&
    get_mismatched_formats() const
    {
        return this->lf_mismatched_formats;
    }

    const std::vector<lnav::console::user_message>& get_format_match_messages()
        const
    {
        return this->lf_format_match_messages;
    }

    struct invalid_line_info {
        static constexpr size_t MAX_INVALID_LINES = 5;

        std::vector<size_t> ili_lines;
        size_t ili_total{0};
    };

    const invalid_line_info& get_invalid_line_info() const
    {
        return this->lf_invalid_lines;
    }

    size_t estimated_remaining_lines() const;

    /**
     * Forget the index entries before the given index for a file opened with
     * loo_streaming.  The last message and enough entries for format
     * detection are always kept, so fewer entries than asked for may be
     * dropped.  Line-number keyed state (opids, thread IDs, pattern locks) is
     * cleared as well.
     *
     * @return The number of entries that were dropped.
     */
    size_t discard_index_before(size_t index);

    /**
     * @return The number of index entries dropped by discard_index_before().
     */
    size_t get_index_base() const { return this->lf_index_base; }

    /**
     * @return The offset of the first line past the end of the time range in
     * the open options, once indexing has reached it.
     */
    std::optional<file_size_t> get_upper_bound_offset() const
    {
        return this->lf_upper_bound_size;
    }

    /**
     * @return The corrections the last rebuild_index() made to the times of
     * the lines before a date rollover, in the order they were made.  Lines
     * that were already dropped with discard_index_before() did not get them.
     */
    const std::vector<time_rollover>& get_time_rollovers() const
    {
        return this->lf_time_rollovers;
    }

    const std::string& get_decompress_error() const
    {
        return this->lf_line_buffer.get_decompress_error();
    }

    time_range get_content_time_range() const;

    const log_level_stats& get_level_stats() const
    {
        return this->lf_level_stats;
    }

protected:
    /**
     * Process a line from the file.
     *
     * @param offset The offset of the line in the file.
     * @param prefix The contents of the line.
     * @param len The length of the 'prefix' string.
     */
    bool process_prefix(shared_buffer_ref& sbr,
                        const line_info& li,
                        scan_batch_context& sbc);

    /**
     * @return Whether anything in the indexing scan still needs this line's
     * bytes.
     *
     * False once no format is going to be found for this file:
     * process_prefix() builds the logline out of the line_info alone on that
     * path, and every block in the scan that does read the line is gated on
     * having a format, on the line being invalid UTF-8, or on a filter being
     * configured.
     *
     * @param observer_wants_text What the logline observer said when the
     * scan started.  Passed in rather than asked for here because the answer
     * holds for the whole pass and this runs on every line.
     */
    bool needs_line_text(const line_info& li, bool observer_wants_text) const;

    void set_base_time_for(scan_batch_context& sbc, const line_info& li);

private:
    logfile(std::filesystem::path filename, const logfile_open_options& loo);

    bool file_options_have_changed();

    void reset_internal_state_for_reindex();

    void reset_time_order();

    void truncate_time_order(size_t line_count);

    void update_time_order(bool recheck);

    bool check_for_timestamp_anchor(scan_batch_context& sbc,
                                    string_fragment sf);

    std::filesystem::path lf_filename;
    std::string lf_filename_as_string;
    logfile_open_options lf_options;
    logfile_activity lf_activity;
    bool lf_named_file{true};
    bool lf_valid_filename{true};
    std::optional<std::filesystem::path> lf_actual_path;
    std::string lf_basename;
    std::string lf_content_id;
    const uint64_t lf_serial{next_serial()};
    struct stat lf_stat{};
    std::shared_ptr<log_format> lf_format;
    log_format_scan_match lf_format_match;
    std::vector<logline> lf_index;
    std::vector<uint32_t> lf_time_order;
    size_t lf_time_order_size{0};
    size_t lf_index_base{0};
    std::chrono::microseconds lf_index_time{0};
    file_off_t lf_index_size{0};
    size_t lf_input_lines{0};
    int lf_index_generation{0};
    stmt_ids lf_stmt_ids;
    line_ids lf_line_ids;
    bool lf_sort_needed{false};
    line_buffer lf_line_buffer;
    int lf_time_offset_line{0};
    timeval lf_time_offset{0, 0};
    bool lf_is_closed{false};
    bool lf_indexing{true};
    line_info lf_last_line_info;
    bool lf_zoned_to_local_state{true};
    robin_hood::unordered_set<string_key, frag_hasher, frag_equal>
        lf_invalidated_opids;
    bool lf_may_have_user_opids{false};
    logline_observer* lf_logline_observer{nullptr};
    index_progress lf_index_progress;
    size_t lf_longest_line{0};
    std::optional<text_format_t> lf_text_format;
    uint32_t lf_out_of_time_order_count{0};
    safe_notes lf_notes;
    std::vector<logline_value_stats> lf_value_stats;
    log_level_stats lf_level_stats;
    pattern_locks lf_pattern_locks;
    /**
     * This file's own timestamp scanner.  The format holds the configuration
     * it is seeded from; the format lock and conversion caches it builds up
     * belong to this file alone.  Mutable because the display side keeps
     * using it as a cache -- reformatting a timestamp locks onto a format and
     * memoizes the local-time offset -- long after indexing is done.
     */
    mutable date_time_scanner lf_time_scanner;
    safe_opid_state lf_opids;
    safe_thread_id_state lf_thread_ids;
    size_t lf_watch_count{0};
    ArenaAlloc::Alloc<char> lf_allocator{64 * 1024};
    std::optional<time_t> lf_cached_base_time;
    std::optional<tm> lf_cached_base_tm;
    anchor_time_t lf_anchor_time{mapbox::util::no_init{}};

    std::optional<std::pair<file_off_t, size_t>> lf_next_line_cache;
    robin_hood::unordered_set<intern_string_t, intern_hasher>
        lf_mismatched_formats;
    /**
     * The set of file types that a format has matched in this file.  Once a
     * format has matched, formats whose file type is not in this set are
     * skipped.
     */
    uint8_t lf_viable_file_types{0};
    bool lf_pruned_formats_logged{false};
    robin_hood::unordered_map<uint32_t, bookmark_metadata> lf_bookmark_metadata;

    std::vector<std::shared_ptr<format_tag_def>> lf_applicable_taggers;
    std::vector<std::shared_ptr<format_partition_def>>
        lf_applicable_partitioners;
    std::map<std::string, metadata> lf_embedded_metadata;
    size_t lf_file_options_generation{0};
    std::optional<std::pair<std::string, lnav::file_options>> lf_file_options;
    std::vector<lnav::console::user_message> lf_format_match_messages;
    invalid_line_info lf_invalid_lines;
    auto_buffer lf_plain_msg_buffer = auto_buffer::alloc(256);
    shared_buffer lf_plain_msg_shared;

    time_range lf_content_time_range;

    struct content_map_entry {
        file_range cme_range;
        std::chrono::microseconds cme_time;
    };
    file_size_t lf_file_size_at_map_time{0};
    std::vector<content_map_entry> lf_content_map;
    std::optional<content_map_entry> lf_lower_bound_entry;
    std::optional<content_map_entry> lf_upper_bound_entry;
    std::optional<file_size_t> lf_upper_bound_size;
    std::vector<time_rollover> lf_time_rollovers;

    struct map_read_upper_bound {};
    struct map_read_lower_bound {
        std::chrono::microseconds mrlb_time;
    };
    using map_read_requirement
        = mapbox::util::variant<map_read_upper_bound, map_read_lower_bound>;

    struct map_entry_not_found {};
    struct map_entry_found {
        content_map_entry mef_entry;
    };
    using map_entry_result
        = mapbox::util::variant<map_entry_not_found, map_entry_found>;

    map_entry_result find_content_map_entry(file_off_t offset,
                                            map_read_requirement req);

    rebuild_result_t build_content_map();
};

class logline_observer {
public:
    virtual ~logline_observer() = default;

    virtual void logline_clear(const logfile& lf) = 0;

    virtual void logline_restart(const logfile& lf, file_size_t rollback_size)
        = 0;

    virtual bool logline_new_lines(const logfile& lf,
                                   logfile::const_iterator ll_begin,
                                   logfile::const_iterator ll_end,
                                   const shared_buffer_ref& sbr) = 0;

    /**
     * @return Whether logline_new_lines() will look at the line text.  Saying
     * no lets the indexer skip reading the bytes and hand over an empty
     * buffer; the observer is still called, since it may have per-line
     * bookkeeping of its own.
     */
    virtual bool logline_wants_text() const { return true; }

    virtual void logline_eof(const logfile& lf) = 0;
};

#endif