Psst. Do you already know segmented logs well? If yes, jump here.

Prologue: The Log 📃🪵

First, let's clarify what we mean by a "log" in this context. Here, log refers to an append only ordered collection of records where the records are ordered by time.

Fig: The Log: An append only ordered collection of records.

Since the records are ordered by time, the log's record indices can be thought of as timestamps, with the convenient property of being decoupled from actual wall-clock time.

The log indices effectively behave as Lamport clock timestamps.

A lamport clock is a logical counter to establish causality between two events. Since it's decoupled from wall-clock time, it's used in distributed-systems for ordering events.

So, why do we care about these kinds of logs? Why are they useful?

Log Usage: An example scenario

There's a teacher and a set of students in a classroom. The teacher wants to hold an elementary arithmetic lesson.

The teacher makes every student write down a particular initial number. (e.g 42). Next, the teacher plans to give a sequence of instructions to the students. The teacher may give the following kinds of instructions:

- Add x; where x is a number
- Sub x; where x is a number
- Mul x; where x is a number
- Div x; where x is a non-zero number

For every instruction, the students are to apply the instruction to their current number, calculate the new number, and write down the new number as their current number. The students are to continue this process for every instruction, till the teacher finishes giving instructions. They are then required to submit the final calculated number to the teacher.

Fig: Sample scenario with initial number 42.

So for instance, if the current number is 42, and the received instruction is Add 3, the student calculates:

result = Add 3 (current_number) = current_number + 3 = 42 + 3 = 45
current_number = result ;; 45

Next if the student receives, Div 5:

result = Div 5 (current_number) = current_number / 5 = 45 / 5 = 9
current_number = result ;; 9

Now, if all students start from the same initial number, and apply the same sequence of operations on their number, they are bound to come to the same result! So the teacher also gives the students a self grading lesson by telling the final number at the end of the class. If the student got the same final number, they scored full marks.

Notice that the students must follow the following to get full marks:

• Start from the same correct initial number
• Apply all the operations correctly in the given order, without random mistakes in the correct pre-determined way.

With computer science, we can model the students as deterministic state machines, the students' current_number as their internal state, and the instructions as inputs to the state machines. From here, we can say the following:

If different identical deterministic state machines start from the same initial state, and receive the same set of inputs in the same order, they will end in the same state.

We call this the state machine replication principle.

State machine replication

Now there's a new problem: the last two rows of students can't hear the teacher properly. What do they do now? The teacher needs a solution that enables them to give the same set of instructions to the backbenchers without actually giving them the answers.

The solution here is to give the instructions in writing. However, in an exercise to foster teamwork between their students, the teacher delegates the instruction sharing task to the students.

Fig: Log based state machine replication between front, middle and back benchers.

The students come up with the following solution:

They first write down the instructions sequentially on sheets of paper, and then perform the calculations separately on their own private sheet. When they are done writing on a sheet of paper with the instructions, they only share the sheet containing the instructions. They never share their private sheet. After passing a sheet full of instructions, they start writing the subsequent instructions on a new sheet of paper.

The backbenchers are able to receive the same set of instructions in the same sequence through the sheets. They perform the necessary calculations on their own private sheet starting from the same initial number using the instructions from the received sheets.

If we inspect carefully, this mechanism of sharing the instructions through the sheets behaves like a log. The private sheets act as the internal states. The collection of sheets collectively act as a log.

Now, in our case, because the backbenchers receive the same set of instructions in the same sequence, they go through the same set of internal states in the same sequence and arrive at the same final state. They effectively replicate the front and middle-benchers. Since, we can model students as state machines, we effectively did state machine replication with a log.

Finally, since the backbenchers receive the instructions through the log and not directly from the teacher, they lag behind a bit but eventually arrive at the same results. So we can say there is a replication lag.

These concepts directly translate to distributed systems. Consider this:

There is a database partition in a distributed database responsible for a certain subset of data. When any data manipulation queries are routed to it, it has to handle the queries. Instead of directly committing the effect of the queries on the underlying storage, it first writes the operations to be applied on the local storage, one-by-one into a log called the "write-ahead-log". Then it applies the operations from the write ahead log to the local storage.

In case there is a database failure, it can re-apply the operations in the write-ahead-log from the last committed entry and arrive at the same state.

When this database partition needs to replicate itself to other follower partitions, it simply replicates the write-ahead-log to the followers instead of copying over the entire materialized state. The followers can use the same write ahead log to arrive to the same state as the leader partition.

Now the follower partitions, have to receive the write-ahead-log first over the network. Only then can they apply the operations. As a result they lag behind the leader. This is replication lag.

Asynchronous processing

Now in the same classroom scenario, what happens when a student is absent? Turns out they still need to do the assignment.

The backbenchers come to the rescue! They share the ordered sheets of instructions and the initial number with the student in distress. The student gleefully applies the instructions on the number, arrives at the same results and scores full marks.

However, notice what happened here: The student did the assignment in a completely out of sync or asynchronous way with respect to the other students.

Logs enable asynchronous processing of requests.

Message queues provide a convenient abstraction over logs to enable asynchronous processing. A server might not be able to synchronously handle all requests due to lack of resources. So instead, it can buffer the requests in a message queue. A different server can then pick up the requests one-by-one and handle them.

Now it's not necessary that all the requests have to be handled by the same server. Because the log is shared, different servers may choose to share the load with each other. In this case, the requests are distributed between the servers to load-balance the requests. (Provided that there is no causal dependency between the requests.)

For instance, a simple scheme might be: If there are N servers, the server for a particular request is decided with request.index % N.

If you want to read more about the usage of logs in distributed systems, read Jay Krep's (co-creator of Apache Kafka) excellent blog post on this topic here.

Segmented Log 🪚🪵

It might come as a surpise, but we have already come across a segmented log in the previous example.

Introduction

In the previous example, the collection of sheets containing the instructions collectively behaved as a log. We can also say that the instruction log was segmented across the different sheets of paper.

Call the individual sheets of paper segments. The collection of sheets can now be called a segmented log.

Let's go back to the log. At the end of the day a log is sequential collection of elements. What's the simplest data structure we can use to implement this?

An array.

However, we need persistence. So let's use a file based abstraction instead.

We can quite literally map a file to a process's virtual memory address space using the mmap() system call and then use it like an array, but that's a topic for a different day.

Fig: A log implementation based on a file.

Since our file based abstraction needs to support an append only data-structure, it internally sequentially writes to the end of the internal file. Assume that this abstraction allows you to uniquely refer to any entry using it's index.

Now, this setup has some problems:

• All entries are sequentially written to a single large file
• A single large file is difficult to store, move and copy
• Few bad sectors in the underlying disk can make the whole file unrecoverable. This can render all stored data unusable.

The logical next step is to split this abstraction across multiple smaller units. We call these smaller units segments.

Fig: segmented_log outline.

In this solution:

• The record index range is split across smaller units called segments. The index ranges of different segments are non-overlapping.
• Each segment individually behaves like a log
• For each segment we maintain an entry: segment. This segment entry stores the index range serviced by it along with a handle to the underlying file
• We keep the segment entries sorted by their starting index
• The first n - 1 segments are called read segments. Their segment entries are stored in a vector called read_segments
• The last segment is called the write segment. We assign it's segment entry to write_segment.

Write behaviour:

• All writes go to the write_segment
• Each segment has a threshold on it's size
• When the write_segment size exceeds it's threshold:
  • We close the write_segment
  • We reopen it as a read segment.
  • We push back the newly opened read segment segment entry to the vector read_segments.
  • We create a new segment with it index range starting after the end of the previous write segment. This segment is assigned to write_segment

Read behaviour (for reading a record at particular index):

• Locate the segment where the index falls within the segment's index range. Look first in the read_segments vector, fall back to write_segment
• Read the record at the given index from the located segment

Original description in the Apache Kafka paper

This section presents the segmented_log as described in the Apache Kafka paper.

Fig: segmented_log (Fig. 2) from the the Apache Kafka paper.

Simple storage: Kafka has a very simple storage layout. Each partition of a topic corresponds to a logical log. Physically, a log is implemented as a set of segment files of approximately the same size (e.g., 1GB). Every time a producer publishes a message to a partition, the broker simply appends the message to the last segment file. For better performance, we flush the segment files to disk only after a configurable number of messages have been published or a certain amount of time has elapsed. A message is only exposed to the consumers after it is flushed.

Unlike typical messaging systems, a message stored in Kafka doesn’t have an explicit message id. Instead, each message is addressed by its logical offset in the log. This avoids the overhead of maintaining auxiliary, seek-intensive random-access index structures that map the message ids to the actual message locations. Note that our message ids are increasing but not consecutive. To compute the id of the next message, we have to add the length of the current message to its id. From now on, we will use message ids and offsets interchangeably.

A consumer always consumes messages from a particular partition sequentially. If the consumer acknowledges a particular message offset, it implies that the consumer has received all messages prior to that offset in the partition. Under the covers, the consumer is issuing asynchronous pull requests to the broker to have a buffer of data ready for the application to consume. Each pull request contains the offset of the message from which the consumption begins and an acceptable number of bytes to fetch. Each broker keeps in memory a sorted list of offsets, including the offset of the first message in every segment file. The broker locates the segment file where the requested message resides by searching the offset list, and sends the data back to the consumer. After a consumer receives a message, it computes the offset of the next message to consume and uses it in the next pull request.

The layout of an Kafka log and the in-memory index is depicted in Figure 2. Each box shows the offset of a message.

The main difference here is that instead of referring to records with a simple index, we refer to it with a logical offset. This is important because the offset is dependent on the record sizes. The offset for next record has to be calculated as the sum of current record offset and current record size.

A segmented_log implementation

The code for this section is in this repository: https://github.com/arindas/laminarmq/ More specifically, in the storage module.

While I would love to discuss testing, benchmarking and profiling, this blog post is becoming quite lengthy. So, please look them up on the repository provided above.

Note: Some of the identifier names might be different on the repository. I have refactored the code sections here to improve readability on various devices. Also there are more comments here to make it easier to understand.

Implementation outline

Fig: Data organisation for persisting a segmented_log on a *nix file system.

A segmented log is a collection of read segments and a single write segment. Each "segment" is backed by a storage file on disk called "store". The offset of the first record in a segment is the base_offset.

The log is:

• "immutable", since only "append", "read" and "truncate" operations are allowed. It is not possible to update or delete records from the middle of the log.
• "segmented", since it is composed of segments, where each segment services records from a particular range of offsets.

All writes go to the write segment. A new record is written at write_segment.next_offset.

When we max out the capacity of the write segment, we close the write segment and reopen it as a read segment. The re-opened segment is added to the list of read segments. A new write segment is then created with base_offset equal to the next_offset of the previous write segment.

When reading from a particular offset, we linearly check which segment contains the given read segment. If a segment capable of servicing a read from the given offset is found, we read from that segment. If no such segment is found among the read segments, we default to the write segment. The following scenarios may occur when reading from the write segment in this case:

• The write segment has synced the messages including the message at the given offset. In this case the record is read successfully and returned.
• The write segment hasn't synced the data at the given offset. In this case the read fails with a segment I/O error.
• If the offset is out of bounds of even the write segment, we return an "out of bounds" error.

Enhancements to the design to enable streaming writes

While the conventional segmented_log data structure is quite performant for a commit_log implementation, it still requires the following properties to hold true for the record being appended:

• We have the entire record in memory
• We know the record bytes' length and record bytes' checksum before the record is appended

It's not possible to know this information when the record bytes are read from an asynchronous stream of bytes. Without the enhancements, we would have to concatenate intermediate byte buffers to a vector. This would not only incur more allocations, but also slow down our system.

Hence, to accommodate this use case, we introduced an intermediate indexing layer to our design.

//! Index and position invariants across segmented_log

// segmented_log index invariants
segmented_log.lowest_index  = segmented_log.read_segments[0].lowest_index
segmented_log.highest_index = segmented_log.write_segment.highest_index

// record position invariants in store
records[i+1].position = records[i].position + records[i].record_header.length

// segment index invariants in segmented_log
segments[i+1].base_index = segments[i].highest_index
                         = segments[i].index[index.len-1].index + 1

Fig: Data organisation for persisting a segmented_log on a *nix file system.

In the new design, instead of referring to records with a raw offset, we refer to them with indices. The index in each segment translates the record indices to raw file position in the segment store file.

Now, the store append operation accepts an asynchronous stream of bytes instead of a contiguously laid out slice of bytes. We use this operation to write the record bytes, and at the time of writing the record bytes, we calculate the record bytes' length and checksum. Once we are done writing the record bytes to the store, we write it's corresponding record_header (containing the checksum and length), position and index as an index_record in the segment index.

This provides two quality of life enhancements:

• Allow asynchronous streaming writes, without having to concatenate intermediate byte buffers
• Records are accessed much more easily with easy to use indices

Now, to prevent a malicious user from overloading our storage capacity and memory with a maliciously crafted request which infinitely loops over some data and sends it to our server, we have provided an optional append_threshold parameter to all append operations. When provided, it prevents streaming append writes to write more bytes than the provided append_threshold.

As we write to segments, the remaining segment capacity is used as the append_threshold. However record bytes aren't guaranteed to be perfectly aligned to segment_capacity.

At the segment level, this requires us to keep a segment_overflow_capacity. All segment append operations now use:

append_threshold = segment_capacity - segment.size + segment_overflow_capacity

A good segment_overflow_capacity value could be segment_capacity / 2.

Component implementation

We now proceed in a bottom-up fashion to implement the entirety of an "indexed segmented log".

AsyncIndexedRead (trait)

If we notice carefully, there is a common thread between Index, Segment and even SegmentedLog as a whole. Even though they are at different levels in the compositional hierarchy, the share some similar traits:

• They allow reading items from specific logical indices
• They have a notion of highest index and lowest index
• The read operation has to be asynchronous in nature to support both in-memory and on-disk storage mechanisms.

Let's formalize these notions:

/// Collection providing asynchronous read access to an indexed set of records (or
/// values).
#[async_trait(?Send)]
pub trait AsyncIndexedRead {
    /// Error that can occur during a read operation.
    type ReadError: std::error::Error;

    /// Value to be read.
    type Value;

    /// Type to index with.
    type Idx: Unsigned + CheckedSub + ToPrimitive + Ord + Copy;

    /// Reads the value at the given index.
    async fn read(&self, idx: &Self::Idx) -> Result<Self::Value, Self::ReadError>;

    /// Index upper exclusive bound
    fn highest_index(&self) -> Self::Idx;

    /// Index lower inclusive bound
    fn lowest_index(&self) -> Self::Idx;

    /// Returns whether the given index is within the index bounds of this
    /// collection.
    ///
    /// This method checks the following condition:
    /// `lowest_index <= idx < highest_index`
    fn has_index(&self, idx: &Self::Idx) -> bool {
        *idx >= self.lowest_index() && *idx < self.highest_index()
    }

    /// Returns the number of values in this collection.
    fn len(&self) -> Self::Idx {
        self.highest_index()
            .checked_sub(&self.lowest_index())
            .unwrap_or(num::zero())
    }

    /// Returns whether this collection is empty.
    fn is_empty(&self) -> bool {
        self.len() == num::zero()
    }

    /// Normalizes the given index between `[0, len)` by subtracting
    /// `lowest_index` from it.
    ///
    /// Returns `Some(normalized_index)` if the index is within
    /// bounds, `None` otherwise.
    fn normalize_index(&self, idx: &Self::Idx) -> Option<Self::Idx> {
        self.has_index(idx)
            .then_some(idx)
            .and_then(|idx| idx.checked_sub(&self.lowest_index()))
    }
}

AsyncTruncate (trait)

Now each of our components need to support a "truncate" operation, where everything sequentially after a certain "mark" is removed. This notion can be expressed as a trait:

/// Trait representing a truncable collection of records, which can be truncated after a
// "mark".
#[async_trait(?Send)]
pub trait AsyncTruncate {
    /// Error that can occur during a truncation operation.
    type TruncError: std::error::Error;

    /// Type to denote a truncation "mark", after which the collection will be truncated.
    type Mark: Unsigned;

    /// Truncates this collection after the given mark, such that this collection
    /// contains records only upto this "mark". (item at "mark" is excluded)
    async fn truncate(&mut self, mark: &Self::Mark) -> Result<(), Self::TruncError>;
}

AsyncConsume (trait)

Next, every abstraction related to storage needs to be safely closed to persist data, or be removed all together. We call such operations "consume" operations. As usual, we codify this general notion with a trait:

/// Trait representing a collection that can be closed or removed entirely.
#[async_trait(?Send)]
pub trait AsyncConsume {
    /// Error that can occur during a consumption operation.
    type ConsumeError: std::error::Error;

    /// Removes all storage associated with this collection.
    ///
    /// The records in this collection are completely removed.
    async fn remove(self) -> Result<(), Self::ConsumeError>;

    /// Closes this collection.
    ///
    /// One would need to re-qcquire a handle to this collection from the storage
    /// in-order to access the records ot this collection again.
    async fn close(self) -> Result<(), Self::ConsumeError>;
}

Sizable (trait)

Every entity capable of storing data needs a mechanism for measuring it's storage footprint i.e size in number of bytes.

/// Trait representing collections which have a measurable size in number of bytes.
pub trait Sizable {
    /// Type to represent the size of this collection in number of bytes.
    type Size: Unsigned + FromPrimitive + Sum + Ord;

    /// Returns the size of this collection in butes.
    fn size(&self) -> Self::Size;
}

Storage (trait)

Finally, we need a mechanism to read and persist data. This mechanism needs to support reads at random positions and appends to the end.

"But Arindam!", you interject. "Such a mechanism exists! It's called a file.", you say triumphantly. And you wouldn't be wrong. However we have the following additional requirements:

• It has to be cross platorm and independent of async runtimes.
• It needs to provide a simple API for random reads without having to seek some pointer.
• It needs to support appending a stream of byte slices.

Alright, let's begin:

#[derive(Debug)]
pub struct StreamUnexpectedLength;

// ... impl Display for StreamUnexpectedLength ...

impl std::error::Error for StreamUnexpectedLength {}

/// Trait representing a read-append-truncate storage media.
#[async_trait(?Send)]
pub trait Storage:
    AsyncTruncate<Mark = Self::Position, TruncError = Self::Error>
    + AsyncConsume<ConsumeError = Self::Error>
    + Sizable<Size = Self::Position>
{
    /// Type to represent the content bytes of this storage media.
    type Content: Deref<Target = [u8]> + Unpin;

    /// Type to represent data positions inside this storage media.
    type Position: Unsigned + FromPrimitive + ToPrimitive + Sum + Ord + Copy;

    /// Error that can occur during storage operations.
    type Error: std::error::Error + From<StreamUnexpectedLength>;

    /// Appends the given slice of bytes to the end of this storage.
    ///
    /// Returns the position at which the slice was written, and the number
    /// of bytes written.
    async fn append_slice(
        &mut self,
        slice: &[u8],
    ) -> Result<(Self::Position, Self::Size), Self::Error>;

    /// Reads `size` number of bytes from the given `position`.
    ///
    /// Returns the bytes read.
    async fn read(
        &self,
        position: &Self::Position,
        size: &Self::Size,
    ) -> Result<Self::Content, Self::Error>;

} // ...where's streaming append?

First, let's unpack what's going on here:

• We support a notion of Content, the slice of bytes that is read
• We also have a notion for Position to identify the position of reads and appends.
• First, we have a simple append_slice() API the simply appends a given slice of bytes to the end of the storage. It returns the position at which slice was written, along with the number of bytes written.
• Next, we have a read() API for reading a slice of bytes of a particular size from a particular position. It returns a Content, the associated type used to represent slice of bytes that are read from this storage.
• Every operation here is fallible. So errors and Result(s) are natural. However, what's up with StreamUnexpectedLength? Keep that in mind for now.
• Our storage can be truncated. We inherit from AsyncTruncate. We treat Position as the truncation mark.
• Our storage is also consumable. We inherit from AsyncConsume for close() and remove()
• Finally, our storage has a notion of size with Sizable. We use our Position type for representing sizes.

Now, we need to support streaming appends with the existing methods.

Let's begin by asking ourselves, what exactly are the arguments here?

A stream of byte slices. How do we represent that?

Let's start with just a slice. Let's call it XBuf. The bound for that is simple enough:

XBuf: Deref<Target = [u8]>

Now we need a stream of these. Also note that we need the reading of a single item from the stream to also be fallible.

First let's just consider a stream of XBuf. Let's call the stream X:

X: Stream<Item = XBuf>

Now, to let's consider an error type XE. To make every read from the stream fallible, X needs the following bounds:

X: Stream<Item = Result<XBuf, XE>>

Now our stream needs to be Unpin so that it we can safely take a &mut reference to it in our function.

This blog post: https://blog.cloudflare.com/pin-and-unpin-in-rust/, goes into detail about why Pin and Unpin are necessary. Also don't forget to consult the standard library documentation:

• pin module: https://doc.rust-lang.org/std/pin/index.html
• Pin struct: https://doc.rust-lang.org/std/pin/struct.Pin.html
• Unpin marker trait: https://doc.rust-lang.org/std/marker/trait.Unpin.html

Apart from our Stream argument, we also need a upper bound on the number of bytes to be written. A Stream can be infinite, but unfortunaly, computer storage is not.

Using the above considerations, let us outline our function:

    // ... inside Storage trait

    async fn append<XBuf, XE, X>(
        &mut self,
        buf_stream: &mut X,
        append_threshold: Option<Self::Size>,
    ) -> Result<(Self::Position, Self::Size), Self::Error>
    where
        XBuf: Deref<Target = [u8]>,
        X: Stream<Item = Result<XBuf, XE>> + Unpin,
    {
        /// ...
    }

When append_threshold is None, we attempt to exhaustively read the entire stream to write to our storage. If it's Some(thresh), we only write upto thresh bytes.

Let's proceed with our implementation:

    // ... inside Storage trait

    async fn append<XBuf, XE, X>(
        &mut self,
        buf_stream: &mut X,
        append_threshold: Option<Self::Size>,
    ) -> Result<(Self::Position, Self::Size), Self::Error>
    where
        XBuf: Deref<Target = [u8]>,
        X: Stream<Item = Result<XBuf, XE>> + Unpin,
    {
        let (mut bytes_written, pos) = (num::zero(), self.size());

        while let Some(buf) = buf_stream.next().await {
             let stream_read_result = match (buf, append_threshold) {
                (Ok(buf), Some(w_cap)) => {
                    match Self::Size::from_usize(buf.deref().len()) {
                        Some(buf_len) if buf_len + bytes_written <= w_cap => Ok(buf),
                        _ => Err::<XBuf, Self::Error>(StreamUnexpectedLength.into()),
                    }
                }
                (Ok(buf), None) => Ok(buf),
                (Err(_), _) => Err(StreamUnexpectedLength.into()),
            };

            // ...
        }
    }

We maintain a counter for the number of bytes already written: bytes_written. For every byte slice read from the stream, we check if it can be accomodated in our storage in accordance with the append_threshold. If not, we error out.

We also keep the write position around in pos. It is simply the size of this storage before we append anything.

Now, let's try to append it:

        // ... inside Storage::append

        while let Some(buf) = buf_stream.next().await {
             let stream_read_result = match (buf, append_threshold) {
                (Ok(buf), Some(w_cap)) => {
                    match Self::Size::from_usize(buf.deref().len()) {
                        Some(buf_len) if buf_len + bytes_written <= w_cap => Ok(buf),
                        _ => Err::<XBuf, Self::Error>(StreamUnexpectedLength.into()),
                    }
                }
                (Ok(buf), None) => Ok(buf),
                (Err(_), _) => Err(StreamUnexpectedLength.into()),
            };

            let append_result = match stream_read_result {
                Ok(buf) => self.append_slice(buf.deref()).await,
                Err(_) => Err(StreamUnexpectedLength.into()),
            };

            // ...
        }

That's reasonable. We append if possible, and propagate the error. Continuing...

        // ... inside Storage::append

        while let Some(buf) = buf_stream.next().await {
             let stream_read_result = match (buf, append_threshold) {
                (Ok(buf), Some(w_cap)) => {
                    match Self::Size::from_usize(buf.deref().len()) {
                        Some(buf_len) if buf_len + bytes_written <= w_cap => Ok(buf),
                        _ => Err::<XBuf, Self::Error>(StreamUnexpectedLength.into()),
                    }
                }
                (Ok(buf), None) => Ok(buf),
                (Err(_), _) => Err(StreamUnexpectedLength.into()),
            };

            let append_result = match stream_read_result {
                Ok(buf) => self.append_slice(buf.deref()).await,
                Err(_) => Err(StreamUnexpectedLength.into()),
            };

            match append_result {
                Ok((_, buf_bytes_w)) => {
                    bytes_written = bytes_written + buf_bytes_w;
                }
                Err(error) => {
                    self.truncate(&pos).await?;
                    return Err(error);
                }
            };
        }

So for every byte slice, we add the number of bytes in it to bytes_written if everything goes well. However, if anything goes wrong:

• We rollback all writes by truncating at the position before all writes, stored in pos.
• We return the error encountered.

Finally, once we exit the loop, we return the position at which the record stream was written, along with the total bytes written:

        } // ... end of: while let Some(buf) = buf_stream.next().await {...}

        Ok((pos, bytes_written))

    } // ... end of Storage::append

We can coalesce the match blocks together by inilining all the results. Putting everything all together:

/// Error to represent undexpect stream termination or overflow, i.e a stream
/// of unexpected length.
#[derive(Debug)]
pub struct StreamUnexpectedLength;

impl std::fmt::Display for StreamUnexpectedLength {
    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
        write!(f, "{self:?}")
    }
}

impl std::error::Error for StreamUnexpectedLength {}

/// Trait representing a read-append-truncate storage media.
#[async_trait(?Send)]
pub trait Storage:
    AsyncTruncate<Mark = Self::Position, TruncError = Self::Error>
    + AsyncConsume<ConsumeError = Self::Error>
    + Sizable<Size = Self::Position>
{
    /// Type to represent the content bytes of this storage media.
    type Content: Deref<Target = [u8]> + Unpin;

    /// Type to represent data positions inside this storage media.
    type Position: Unsigned + FromPrimitive + ToPrimitive + Sum + Ord + Copy;

    /// Error that can occur during storage operations.
    type Error: std::error::Error + From<StreamUnexpectedLength>;

    /// Appends the given slice of bytes to the end of this storage.
    ///
    /// Implementations must update internal cursor or write pointers, if any,
    /// when implementing this method.
    async fn append_slice(
        &mut self,
        slice: &[u8],
    ) -> Result<(Self::Position, Self::Size), Self::Error>;

    /// Appends a stream of byte slices to the end of this storage.
    ///
    /// This method writes at max `append_threshold` number of bytes from the
    /// given stream of bytes slices. If the provided `append_threshold` is
    /// `None`, no such check is enforced; we attempt to write the entire
    /// stream until it's exhausted.
    ///
    /// The following error scenarios may occur during writing:
    /// - `append_threshold` is `Some(_)`, and the stream contains more bytes
    /// than the threshold
    /// - The stream unexpectedly yields an error when attempting to read the
    /// next byte slice from the stream
    /// - There is a storage error when attempting to write one of the byte
    /// slices from the stream.
    ///
    /// In all of the above error cases, we truncate this storage media with
    /// the size of the storage media before we started the append operation,
    /// effectively rolling back any writes.
    ///
    /// Returns the position where the bytes were written and the number of
    /// bytes written.
    async fn append<XBuf, XE, X>(
        &mut self,
        buf_stream: &mut X,
        append_threshold: Option<Self::Size>,
    ) -> Result<(Self::Position, Self::Size), Self::Error>
    where
        XBuf: Deref<Target = [u8]>,
        X: Stream<Item = Result<XBuf, XE>> + Unpin,
    {
        let (mut bytes_written, pos) = (num::zero(), self.size());

        while let Some(buf) = buf_stream.next().await {
            match match match (buf, append_threshold) {
                (Ok(buf), Some(w_cap)) => {
                    match Self::Size::from_usize(buf.deref().len()) {
                        Some(buf_len) if buf_len + bytes_written <= w_cap => Ok(buf),
                        _ => Err::<XBuf, Self::Error>(StreamUnexpectedLength.into()),
                    }
                }
                (Ok(buf), None) => Ok(buf),
                (Err(_), _) => Err(StreamUnexpectedLength.into()),
            } {
                Ok(buf) => self.append_slice(buf.deref()).await,
                Err(_) => Err(StreamUnexpectedLength.into()),
            } {
                Ok((_, buf_bytes_w)) => {
                    bytes_written = bytes_written + buf_bytes_w;
                }
                Err(error) => {
                    self.truncate(&pos).await?;
                    return Err(error);
                }
            };
        }

        Ok((pos, bytes_written))
    }

    /// Reads `size` number of bytes from the given `position`.
    ///
    /// Returns the bytes read.
    async fn read(
        &self,
        position: &Self::Position,
        size: &Self::Size,
    ) -> Result<Self::Content, Self::Error>;
}

Now to answer one of the initial questions, we needed StreamUnexpectedLength as a sentinel error type to represent the error case when the stream unexpectedly errors out while reading, or has more bytes in total than our append_threshold.

A sample Storage impl.

Let's explore a tokio::fs::File based implementation of Storage:

First, let's outline our struct:

pub struct StdSeekReadFileStorage {
    storage: RwLock<BufWriter<TokioFile>>,
    backing_file_path: PathBuf,

    size: u64,
}

TokioFile is an alias for tokio::fs::File. It's defined in a use directive in the source.

We need a buffered writer over our file to avoid hitting the file too many times for small writes. Since our workload will largely be composed of small writes and reads, this is important.

The need for RwLock will be clear in a moment.

Next, let's proceed with the constructor for this struct:

impl StdSeekReadFileStorage {
    pub async fn new<P: AsRef<Path>>(path: P) -> Result<Self, StdSeekReadFileStorageError> {
        let backing_file_path = path.as_ref().to_path_buf();

        let storage = Self::obtain_backing_storage(&backing_file_path).await?;

        let initial_size = storage
            .metadata()
            .await
            .map_err(StdSeekReadFileStorageError::IoError)?
            .len();

        Ok(Self {
            storage: RwLock::new(BufWriter::new(storage)),
            backing_file_path,
            size: initial_size,
        })
    }

    async fn obtain_backing_storage<P: AsRef<Path>>(
        path: P,
    ) -> Result<TokioFile, StdSeekReadFileStorageError> {
        OpenOptions::new()
            .write(true)
            .append(true)
            .create(true)
            .read(true)
            .open(path)
            .await
            .map_err(StdSeekReadFileStorageError::IoError)
    }
}

As you can see, for the underlying storage file, we enable the following flags:

• write: enables writing to the file
• append: new writes are appended (as opposed to truncating the file before writes)
• create: if the file doesn't exist, it's created
• read: enable reading from the file

Now before we implement the Storage trait for StdSeekReadFileStorage, we need to implement Storage's inherited traits. Let's proceed one by one.

First, we have Sizable:

impl Sizable for StdSeekReadFileStorage {
    type Size = u64;

    fn size(&self) -> Self::Size {
        self.size
    }
}

Next, we implement AsyncTruncate:

#[async_trait(?Send)]
impl AsyncTruncate for StdSeekReadFileStorage {
    type Mark = u64;

    type TruncError = StdSeekReadFileStorageError;

    async fn truncate(&mut self, position: &Self::Mark) -> Result<(), Self::TruncError> {
        // before truncating, flush all writes
        self.storage
            .write()
            .await
            .flush()
            .await
            .map_err(StdSeekReadFileStorageError::IoError)?;

        // reopen file directly for truncation
        let writer = Self::obtain_backing_storage(&self.backing_file_path).await?;

        // truncate at the given position
        writer
            .set_len(*position)
            .await
            .map_err(StdSeekReadFileStorageError::IoError)?;

        // close old file handle and assign thr new file handle to storage
        self.storage = RwLock::new(BufWriter::new(writer));
        self.size = *position; // update size after truncation

        Ok(())
    }
}

We also need AsyncConsume:

#[async_trait(?Send)]
impl AsyncConsume for StdSeekReadFileStorage {
    type ConsumeError = StdSeekReadFileStorageError;

    async fn remove(mut self) -> Result<(), Self::ConsumeError> {
        let backing_file_path = self.backing_file_path.clone();

        self.close().await?;

        tokio::fs::remove_file(&backing_file_path)
            .await
            .map_err(StdSeekReadFileStorageError::IoError)
    }

    async fn close(mut self) -> Result<(), Self::ConsumeError> {
        self.storage
            .write()
            .await
            .flush()
            .await
            .map_err(StdSeekReadFileStorageError::IoError)
    }
}

With the pre-requisites ready, let's proceed with our Storage impl:

#[async_trait(?Send)]
impl Storage for StdSeekReadFileStorage {
    type Content = Vec<u8>;

    type Position = u64;

    type Error = StdSeekReadFileStorageError;

    async fn append_slice(
        &mut self,
        slice: &[u8],
    ) -> Result<(Self::Position, Self::Size), Self::Error> {
        let current_position = self.size;

        // write to storage using he BufWriter
        self.storage
            .write()
            .await
            .write_all(slice)
            .await
            .map_err(StdSeekReadFileStorageError::IoError)?;

        let bytes_written = slice.len() as u64;

        self.size += bytes_written; // update storage size

        Ok((current_position, bytes_written))
    }

    async fn read(
        &self,
        position: &Self::Position,
        size: &Self::Size,
    ) -> Result<Self::Content, Self::Error> {
        // validate that the record to be read is within the written area
        if *position + *size > self.size {
            return Err(StdSeekReadFileStorageError::ReadBeyondWrittenArea);
        }

        // pre-allocate buffer to use for reading
        let mut read_buf = vec![0_u8; *size as usize];

        // acquire &mut storage from behind &self using RwLock::write
        let mut storage = self.storage.write().await;

        // seek to read position
        storage
            .seek(io::SeekFrom::Start(*position))
            .await
            .map_err(StdSeekReadFileStorageError::IoError)?;

        // read required number of bytes
        storage
            .read_exact(&mut read_buf)
            .await
            .map_err(StdSeekReadFileStorageError::IoError)?;

        // seek to end of file using current size of file
        storage
            .seek(io::SeekFrom::Start(self.size))
            .await
            .map_err(StdSeekReadFileStorageError::IoError)?;

        Ok(read_buf)
    }
}

Notice that the Storage trait uses a &self for Storage::read. This was to support idempotent operations.

In our case, we need to update the seek position of the file, from behind a &self. So we need some interior mutability to achieve this. Hence the RwLock.

Hopefully, the need for the RwLock is clear now. In retrospect, we could also have used a Mutex but using a RwLock keeps the option of multiple readers for read only operations open.

Note that, the read operation is still idempotent as we restore the old file position after reading.

Record (struct)

Before we move on to the concept of a CommitLog, we need to abstract a much more fundamental aspect of our implementation. How do we represent the actual "records"?

Let's see... a record in the most general sense needs only two things:

• The actual value to be contained in the record
• Some metadata about the record

So, we express that directly:

pub struct Record<M, T> {
    pub metadata: M,
    pub value: T,
}

CommitLog (trait)

Now, with the abstractions presented above, we are ready to express the notion of a CommitLog. The properties of a CommitLog are:

1. It allows reading records from random indices
2. Naturally it has some index bounds
3. It allows appending records which may contain a stream of byte slices as it value.
4. It can be truncated at a specific index.
5. It supports close() and remove() operations to safely persist or remove data respectively.

All of these properties have already been represented with the traits above. We now use them to define the concept of a CommitLog:

#[async_trait::async_trait(?Send)]
pub trait CommitLog<M, T>:
    AsyncIndexedRead<Value = Record<M, T>, ReadError = Self::Error> // (1, 2)
    + AsyncTruncate<Mark = Self::Idx, TruncError = Self::Error> // (4)
    + AsyncConsume<ConsumeError = Self::Error> // (5)
    + Sizable // Of course, we need to know the total storage size
{
    /// Associated error type for fallible operations
    type Error: std::error::Error;

    // (3)
    async fn append<X, XBuf, XE>(&mut self, record: Record<M, X>) -> Result<Self::Idx, Self::Error>
    where
        X: futures_lite::Stream<Item = Result<XBuf, XE>>,
        X: Unpin + 'async_trait,
        XBuf: std::ops::Deref<Target = [u8]>;

    /// Removes expired records i.e records older than the given _expiry_duration.
    ///
    /// The default implementation doesn't remove any records.
    ///
    /// Returns the number of records removed.
    async fn remove_expired(
        &mut self,
        _expiry_duration: std::time::Duration,
    ) -> Result<Self::Idx, Self::Error> {
        // remove zero records by default
        async { Ok(<Self::Idx as num::Zero>::zero()) }.await
    } // ... what's this?
}

Optionally, a CommitLog implementation might need to remove some records that are older by a certain measure of time. Let's call them expired records. So we provide a function for that in case different implementations need it.

Index (struct)

Let's start with our first direct component of our indexed segmented log, the Index.

First, we need to answer two primary questions:

• What kind of data are we storing?
• In what layout will we store the said data?

So recall, at a high level, an Index is logical mapping from indices to byte positions on storage.

So this at least has store 2-tuples of the form: (record_index, record_position)

Now we need two additional data points:

• Number of bytes in the record i.e record_length
• A checksum of the contents of the record, e.g. crc32

These two datapoints are essential to verify if the record data is valid or corrupted on the storage media. ("storage media" = Storage trait impl.)

So we arrive at this 4-tuple: (checksum, length, index, position)

Let's call this an IndexRecord.

Now, an Index stores index records sequentially. So:

index_record[i+1].index = index_record[i].index + 1

Now, all these 4-tuples need the exact same number of bytes to be stored. Let's call this size IRSZ. If the index records are laid out sequentially:, every index record will be at a position which is an integral multiple of IRSZ:

## storage (Index)

(checksum, length, index, position) @ storage::position = 0
(checksum, length, index, position) @ storage::position = 1 x IRSZ
(checksum, length, index, position) @ storage::position = 2 x IRSZ
(checksum, length, index, position) @ storage::position = 3 x IRSZ
...

Note: the position in the tuple refers to the position of the actual Record in Store. storage::position here refers to the position within the Index file (Storage impl).

Due to this property, the index can be derived from the position of the record itself. The number of records is simply:

len(Index) = size(Index) / IRSZ

Using this, we can conclude that storing the index in each IndexRecord is redundant.

However, an Index can start from an arbitrary high index. Let's call this the base_index. So if we store a marker record of sorts, the contains the base_index, and then store all the index records after it sequentially, we can say:

// simple row major address calculation
index_record_position_i = size(base_marker) + i * IRSZ

So now we can lay out our IndexRecord instances on storage as follows:

## storage (Index)

[base_index_marker]          @ storage::position = 0
(checksum, length, position) @ storage::position = size(base_index_marker) + 0
(checksum, length, position) @ storage::position = size(base_index_marker) + 1 x IRSZ
(checksum, length, position) @ storage::position = size(base_index_marker) + 2 x IRSZ
(checksum, length, position) @ storage::position = size(base_index_marker) + 3 x IRSZ
...

Now, number of records is calculated as:

// number of records in Index
len(Index) = (size(Index) - size(base_index_marker)) / IRSZ

Now let's finalize the bytewise layout on storage:

## IndexBaseMarker (size = 16 bytes)
┌───────────────────────────┬──────────────────────────┐
│ base_index: u64 ([u8; 8]) │ _padding:  u64 ([u8; 8]) │
└───────────────────────────┴──────────────────────────┘

## IndexRecord (size = 16 bytes)
┌─────────────────────────┬───────────────────────┬─────────────────────────┐
│ checksum: u64 ([u8; 8]) │ length: u32 ([u8; 4]) │ position: u32 ([u8; 4]) │
└─────────────────────────┴───────────────────────┴─────────────────────────┘

We add padding to the IndexBaseMarker to keep it aligned with IndexRecord.

We represent these records as follows:

pub struct IndexBaseMarker {
    pub base_index: u64,
    _padding: u64,
}

#[derive(Debug, Default, Clone, Copy, PartialEq, Eq)]
pub struct IndexRecord {
    pub checksum: u64,
    pub length: u32,
    pub position: u32,
}

Since we 32 bytes to represent positions of records, a Segment can contain only be upto 4GiB (2³² unique byte positions = 2³² bytes = 4GiB). Practically speaking, segments generally don't exceed 1GB. If segments are too big, individual segments are difficult to move around. So this limit is not a problem.

We use binary encoding to store these records.

Now we could use serde and bincode to serialize these records on Storage impls. However, since these records will be serialized and deserialized fairly often, I wanted to serialize in constant space, with a simple API.

First, let us generalize over both IndexBaseMarker and IndexRecord. We need to formalize an entity with the folowing properties:

• It has a known size at compile time
• It can be read from and written to any storage

We can express this directly:

trait SizedRecord: Sized {
    fn write<W: Write>(&self, dest: &mut W) -> std::io::Result<()>;

    fn read<R: Read>(source: &mut R) -> std::io::Result<Self>;
}

Now we need a kind of SizedRecord that can be stored on a Storage impl. Let's call it PersistentSizedRecord:

/// Wrapper struct to enable `SizedRecord` impls to be stored on Storage impls.
///
/// REPR_SIZE is the number of bytes required to store the inner SizedRecord.
struct PersistentSizedRecord<SR, const REPR_SIZE: usize>(SR);

impl<SR, const REPR_SIZE: usize> PersistentSizedRecord<SR, REPR_SIZE> {
    fn into_inner(self) -> SR {
        self.0
    }
}

impl<SR: SizedRecord, const REPR_SIZE: usize> PersistentSizedRecord<SR, REPR_SIZE> {
    async fn read_at<S>(source: &S, position: &S::Position) -> Result<Self, IndexError<S::Error>>
    where
        S: Storage,
    {
        // read exactly REPR_SIZE bytes from source Storage impl. at the given position
        let record_bytes = source
            .read(
                position,
                &<S::Size as FromPrimitive>::from_usize(REPR_SIZE)
                    .ok_or(IndexError::IncompatibleSizeType)?,
            )
            .await
            .map_err(IndexError::StorageError)?; // read bytes for record

        let mut cursor = Cursor::new(record_bytes.deref()); // wrap to a Read impl.

        SR::read(&mut cursor).map(Self).map_err(IndexError::IoError) // deserialize
    }

    async fn append_to<S>(&self, dest: &mut S) -> Result<S::Position, IndexError<S::Error>>
    where
        S: Storage,
    {
        let mut buffer = [0_u8; REPR_SIZE]; // buffer to store Serialized record
        let mut cursor = Cursor::new(&mut buffer as &mut [u8]); // wrap to a Write impl.

        self.0.write(&mut cursor).map_err(IndexError::IoError)?; // serialize

        let (position, _) = dest
            .append_slice(&buffer)
            .await
            .map_err(IndexError::StorageError)?; // append to storage

        Ok(position)
    }
}

Next, we implement the SizedRecord trait for IndexBaseMarker and IndexRecord:

impl SizedRecord for IndexBaseMarker {
    fn write<W: Write>(&self, dest: &mut W) -> std::io::Result<()> {
        dest.write_u64::<LittleEndian>(self.base_index)?;

        Ok(())
    }

    fn read<R: Read>(source: &mut R) -> std::io::Result<Self> {
        let base_index = source.read_u64::<LittleEndian>()?;

        Ok(Self {
            base_index,
            _padding: 0_u64,
        })
    }
}

impl SizedRecord for IndexRecord {
    fn write<W: Write>(&self, dest: &mut W) -> std::io::Result<()> {
        dest.write_u64::<LittleEndian>(self.checksum)?;
        dest.write_u32::<LittleEndian>(self.length)?;
        dest.write_u32::<LittleEndian>(self.position)?;

        Ok(())
    }

    fn read<R: Read>(source: &mut R) -> std::io::Result<Self> {
        let checksum = source.read_u64::<LittleEndian>()?;
        let length = source.read_u32::<LittleEndian>()?;
        let position = source.read_u32::<LittleEndian>()?;

        Ok(IndexRecord {
            checksum,
            length,
            position,
        })
    }
}

[ Quiz 💡]: We dont read or write the _padding bytes in our IndexBaseMarker SizedRecord impl. So how is it still aligned?

[ A ]: Remember that we pass in a const generic parameter REPR_SIZE when creating a PersistentSizedRecord. When writing or reading, we always read REPR_SIZE number of bytes, regardless of how we serialize or deserialize our IndexRecord or IndexBaseMarker. In this case we just pass a const usize with value 16.

We also declare some useful constants to keep things consistent:

/// Extension used by backing files for Index instances.
pub const INDEX_FILE_EXTENSION: &str = "index";

/// Number of bytes required for storing the base marker.
pub const INDEX_BASE_MARKER_LENGTH: usize = 16;

/// Number of bytes required for storing the record header.
pub const INDEX_RECORD_LENGTH: usize = 16;

/// Lowest underlying storage position
pub const INDEX_BASE_POSITION: u64 = 0;

Before we proceed with our Index implementation, let us do a quick back of the handle estimate on how big Index files can be.

Every IndexRecord is 16 bytes. So for every Record we have 16 bytes. Let's assume that Record sizes are 1KB on average. Let's assume that Segment files are 1GB on average.

So we can calculate as follows:

            1GB Segment == pow(10, 6) * 1KB Record

         1 * 1KB Record =>          1 * 16B IndexRecord

pow(10, 6) * 1KB Record => pow(10, 6) * 16B IndexRecord
                        => 16MB Index

Or,         1GB Segment => 16MB Index (Result)


e.g. 10 * 1GB segment files => 10 * 16MB Index files = 160 MB overhead

     ITB total data through 1000 segment files => 16GB overhead

Keep this calculation in mind as we proceed through our implementation.

With the groundwork ready, let's begin our Index implementation:

pub struct Index<S, Idx> {
    /// In memory cache of IndexRecord instaqnces
    index_records: Option<Vec<IndexRecord>>,

    base_index: Idx, /// Index stored in the first IndexRecord
    next_index: Idx, /// Index to be stored in the next IndexRecord to be added

    storage: S,      /// Underlying storage for Index
}

Why do we need an in-memory cache in the fist place? Well IndexRecord instances are fairly small and there are usually few of them (<= 1000) in an Index. A simple in-memory cache makes sense rather than hitting the storage everytime. (We could probably mmap() but this is simple enough.)

Alright then, why make it optional?

Recall that for 1TB of data with 1KB record size, we end up having 16GB of Index overhead. It's clearly not practical allocating this amount of memory, since we expect our system to able to handle this scale.

So we make caching IndexRecord instances optional. This would enable us to decide which Index instances to cache based on access patterns.

For instance, we could maintain an LRUCache of Index instances that are currently cached. When an Index outside of the LRUCache is accessed, we add it to the LRUCache. When an Index from within the LRUCache is accessed, we update the LRUCache accordingly. The LRUCache will have some maximum capacity, which decides the maximum number of Index instances that can be cached at the same time. We could replace LRUCache with other kinds of cache (e.g. LFUCache) for different performance characteristics. The Index files are still persisted on storage so there is no loss of data.

I wanted this implementation to handle 1TB of data on a Raspberry Pi 3B. Unfortunately, it has only 1GB RAM. However, if we enforce a limit that only 10 Index instances are cached at a time (e.g. by setting the LRUCache max capacity to 10), that would be a 160MB overhead. That would make this implementation usable on an RPi 3B, albeit at the cost of some latency.

For storage, I can connect a 1TB external hard disk to the RPi 3B and proceed as usual.

Now, let's define some utilities for constructing Index instances.

impl<S, Idx> Index<S, Idx>
where
    S: Storage,
    Idx: Unsigned + FromPrimitive + Copy + Eq,
{

    /// Estimates the numer of IndexRecord instances in the given Storage impl ref.
    ///
    /// Returns the number of IndexRecord instances estimated to be present.
    pub fn estimated_index_records_len_in_storage(
        storage: &S,
    ) -> Result<usize, IndexError<S::Error>> {
        let index_storage_size = storage // size of the given storage for Index
            .size()
            .to_usize()
            .ok_or(IndexError::IncompatibleSizeType)?;

        // len(Index) = (size(Index) - size(IndexBaseMarker)) / size(IndexRecord)
        let estimated_index_records_len =
            index_storage_size.saturating_sub(INDEX_BASE_MARKER_LENGTH) / INDEX_RECORD_LENGTH;

        Ok(estimated_index_records_len)
    }

    /// Obtains the base_index by reading the IndexBaseMarker from the given storage.
    ///
    /// Returns the base index.
    pub async fn base_index_from_storage(storage: &S) -> Result<Idx, IndexError<S::Error>> {
        // read the index base marker from the base position (0)
        let index_base_marker =
            PersistentSizedRecord::<IndexBaseMarker, INDEX_BASE_MARKER_LENGTH>::read_at(
                storage,
                &u64_as_position!(INDEX_BASE_POSITION, S::Position)?,
            )
            .await
            .map(|x| x.into_inner());

        // map Result<IndexBaseMarker, ...> to Result<Idx, ...>
        index_base_marker
            .map(|x| x.base_index)
            .and_then(|x| u64_as_idx!(x, Idx))
    }

    /// Reads the IndexRecord instances present in the given Storage impl ref.
    ///
    /// Returns a Vec of IndexRecord instances.
    pub async fn index_records_from_storage(
        storage: &S,
    ) -> Result<Vec<IndexRecord>, IndexError<S::Error>> {
        // start reading after the IndexBaseMarker
        let mut position = INDEX_BASE_MARKER_LENGTH as u64;

        let estimated_index_records_len = Self::estimated_index_records_len_in_storage(storage)?;

        // preallocate the vector for storing IndexRecord instances
        let mut index_records = Vec::<IndexRecord>::with_capacity(estimated_index_records_len);

        // while index records can be read without error
        while let Ok(index_record) =
            PersistentSizedRecord::<IndexRecord, INDEX_RECORD_LENGTH>::read_at(
                storage,
                &u64_as_position!(position, S::Position)?,
            )
            .await
        {
            // append the IndexRecord read just now to the Vec<IndexRecord>
            index_records.push(index_record.into_inner());

            // advance to the next position for reading an IndexRecord
            position += INDEX_RECORD_LENGTH as u64;
        }

        index_records.shrink_to_fit(); // release empty space left, if any

        // cross verify the number of index records read
        if index_records.len() != estimated_index_records_len {
            Err(IndexError::InconsistentIndexSize)
        } else {
            Ok(index_records)
        }
    }

    /// Cross validates the given base index against the one in the storage.
    ///
    /// Returns the validated base index.
    pub async fn validated_base_index(
        storage: &S,
        base_index: Option<Idx>,
    ) -> Result<Idx, IndexError<S::Error>> {
        // read the base index from the given storage
        let read_base_index = Self::base_index_from_storage(storage).await.ok();

        match (read_base_index, base_index) {
            // a. error out if neither in storage, nor provided
            (None, None) => Err(IndexError::NoBaseIndexFound),

            // b. either only in storage or only provided, choose what is present
            (None, Some(base_index)) => Ok(base_index),
            (Some(base_index), None) => Ok(base_index),

            // c. present in both storage, as well as provided

            // c.1. conflicting, error out
            (Some(read), Some(provided)) if read != provided => Err(IndexError::BaseIndexMismatch),

            // c.2. no conflict, choose provided (no difference)
            (Some(_), Some(provided)) => Ok(provided),
        }
    }

    // ...
}

Next, we define our constructors for Index:

impl<S, Idx> Index<S, Idx>
where
    S: Storage,
    Idx: Unsigned + FromPrimitive + Copy + Eq,
{
    // ...

    /// Creates an Index instance from a `Storage` impl instance and an optional base index.
    ///
    /// Reads the IndexRecord instances present in the given storage and caches them.
    ///
    /// Returns an Index instance.
    pub async fn with_storage_and_base_index_option(
        storage: S,
        base_index: Option<Idx>,
    ) -> Result<Self, IndexError<S::Error>> {
        // cross validates the given base index with the one present on storage.
        let base_index = Self::validated_base_index(&storage, base_index).await?;

        // reads the IndexRecord instances preent in the provided storage
        let index_records = Self::index_records_from_storage(&storage).await?;

        let len = index_records.len() as u64;

        let next_index = base_index + u64_as_idx!(len, Idx)?;

        Ok(Self {
            index_records: Some(index_records),
            base_index,
            next_index,
            storage,
        })
    }

    pub async fn with_storage_and_base_index(
        storage: S,
        base_index: Idx,
    ) -> Result<Self, IndexError<S::Error>> {
        Self::with_storage_and_base_index_option(storage, Some(base_index)).await
    }

    pub async fn with_storage(storage: S) -> Result<Self, IndexError<S::Error>> {
        Self::with_storage_and_base_index_option(storage, None).await
    }


    /// Creates an Index with the given storage, cached index records and a
    /// validated base index.
    ///
    /// This function doesn't touch the provided Storage impl instance,
    /// other than reading its size. The cached index record instances
    /// are used as is. If no cache if provided, then the created Index
    /// is not cached.
    ///
    /// This function is primarily useful when flushing an Index instance by
    /// closing the underlying storage and re-opening it, without reading all
    /// the IndexRecord instances again.
    pub fn with_storage_index_records_option_and_validated_base_index(
        storage: S,
        index_records: Option<Vec<IndexRecord>>,
        validated_base_index: Idx,
    ) -> Result<Self, IndexError<S::Error>> {
        let len = Self::estimated_index_records_len_in_storage(&storage)? as u64;
        let next_index = validated_base_index + u64_as_idx!(len, Idx)?;

        Ok(Self {
            index_records,
            base_index: validated_base_index,
            next_index,
            storage,
        })
    }

    // ...
}

Next, we define some functions for managing caching behaviour:

impl<S, Idx> Index<S, Idx>
where
    S: Storage,
    Idx: Unsigned + FromPrimitive + Copy + Eq,
{
    // ...

    /// Takes the cached index records from the Index, leaving it uncached.
    pub fn take_cached_index_records(&mut self) -> Option<Vec<IndexRecord>> {
        self.index_records.take()
    }

    pub fn cached_index_records(&self) -> Option<&Vec<IndexRecord>> {
        self.index_records.as_ref()
    }

    /// Caches this Index if not already cached.
    pub async fn cache(&mut self) -> Result<(), IndexError<S::Error>> {
        if self.index_records.as_ref().is_some() {
            return Ok(());
        }

        self.index_records = Some(Self::index_records_from_storage(&self.storage).await?);

        Ok(())
    }
}

Some minor utilities for ease of implementation:

impl<S, Idx> Index<S, Idx>
where
    S: Default,
    Idx: Copy,
{
    pub fn with_base_index(base_index: Idx) -> Self {
        Self {
            index_records: Some(Vec::new()),
            base_index,
            next_index: base_index,
            storage: S::default(),
        }
    }
}

impl<S: Storage, Idx> Sizable for Index<S, Idx> {
    type Size = S::Size;

    fn size(&self) -> Self::Size {
        self.storage.size()
    }
}

impl<S: Storage, Idx> Index<S, Idx> {

    /// Returns the position for the IndexRecord corresponding to the given
    /// normalized_index on the underlying storage.
    #[inline]
    fn index_record_position(normalized_index: usize) -> Result<S::Position, IndexError<S::Error>> {
        let position = (INDEX_BASE_MARKER_LENGTH + INDEX_RECORD_LENGTH * normalized_index) as u64;
        u64_as_position!(position, S::Position)
    }
}

impl<S, Idx> Index<S, Idx>
where
    S: Storage,
    Idx: Unsigned + CheckedSub + ToPrimitive + Ord + Copy,
{

    /// Maps the given Idx to a usize in [0, len(Index))
    #[inline]
    fn internal_normalized_index(&self, idx: &Idx) -> Result<usize, IndexError<S::Error>> {
        self.normalize_index(idx)
            .ok_or(IndexError::IndexOutOfBounds)?
            .to_usize()
            .ok_or(IndexError::IncompatibleIdxType)
    }
}

Now, we move on to the primary responsiblities of our Index.

First, let's implement a mechanism to read IndexRecord instances from our Index:

#[async_trait::async_trait(?Send)]
impl<S, Idx> AsyncIndexedRead for Index<S, Idx>
where
    S: Storage,
    Idx: Unsigned + CheckedSub + ToPrimitive + Ord + Copy,
{
    type ReadError = IndexError<S::Error>;

    type Idx = Idx;

    /// The type of value to be read
    type Value = IndexRecord;

    fn highest_index(&self) -> Self::Idx {
        self.next_index
    }

    fn lowest_index(&self) -> Self::Idx {
        self.base_index
    }

    /// Reads the IndexRecord corresponding to the given idx.
    async fn read(&self, idx: &Self::Idx) -> Result<Self::Value, Self::ReadError> {
        let normalized_index = self.internal_normalized_index(idx)?;

        // If this index is cached, read from the cached Vec of IndexRecord instances
        if let Some(index_records) = self.index_records.as_ref() {
            index_records
                .get(normalized_index)
                .ok_or(IndexError::IndexGapEncountered)
                .map(|&x| x)
        } else { // otherwise, read from the underlying storage
            PersistentSizedRecord::<IndexRecord, INDEX_RECORD_LENGTH>::read_at(
                &self.storage,
                &Self::index_record_position(normalized_index)?,
            )
            .await
            .map(|x| x.into_inner())
        }
    }
}

Next, we need a mechanism to append IndexRecord instances to our Index:

impl<S, Idx> Index<S, Idx>
where
    S: Storage,
    Idx: Unsigned + ToPrimitive + Copy,
{
    pub async fn append(&mut self, index_record: IndexRecord) -> Result<Idx, IndexError<S::Error>> {
        let write_index = self.next_index;

        // If this Index is empty, we need to write the IndexBaseMarker first before
        // writing any IndexRecord
        if write_index == self.base_index {
            PersistentSizedRecord::<IndexBaseMarker, INDEX_BASE_MARKER_LENGTH>(
                IndexBaseMarker::new(idx_as_u64!(write_index, Idx)?),
            )
            .append_to(&mut self.storage)
            .await?;
        }

        // Append the IndexRecord to storage
        PersistentSizedRecord::<IndexRecord, INDEX_RECORD_LENGTH>(index_record)
            .append_to(&mut self.storage)
            .await?;

        // If this Ineex is cached, also append the IndexRecord to the cache
        if let Some(index_records) = self.index_records.as_mut() {
            index_records.push(index_record);
        }

        self.next_index = write_index + Idx::one(); // update the next_index
        Ok(write_index)
    }
}

We need to be able to truncate our Index:

#[async_trait::async_trait(?Send)]
impl<S, Idx> AsyncTruncate for Index<S, Idx>
where
    S: Storage,
    Idx: Unsigned + CheckedSub + ToPrimitive + Ord + Copy,
{
    type TruncError = IndexError<S::Error>;

    type Mark = Idx;

    /// Truncates this Index at the given Index. IndexRecord instances at
    /// indices >= idx are removed.
    async fn truncate(&mut self, idx: &Self::Mark) -> Result<(), Self::TruncError> {
        let normalized_index = self.internal_normalized_index(idx)?;

        self.storage
            .truncate(&Self::index_record_position(normalized_index)?)
            .await
            .map_err(IndexError::StorageError)?;

        if let Some(index_records) = self.index_records.as_mut() {
            index_records.truncate(normalized_index);
        }

        self.next_index = *idx;

        Ok(())
    }
}

Finally, we define how to close or remove our Index:

#[async_trait::async_trait(?Send)]
impl<S: Storage, Idx> AsyncConsume for Index<S, Idx> {
    type ConsumeError = IndexError<S::Error>;

    async fn remove(self) -> Result<(), Self::ConsumeError> {
        self.storage
            .remove()
            .await
            .map_err(IndexError::StorageError)
    }

    async fn close(self) -> Result<(), Self::ConsumeError> {
        self.storage.close().await.map_err(IndexError::StorageError)
    }
}

Notice how all of the primary functions of our Index are supported by the traits we wrote earlier.

Store (struct)

Now that we have our Index ready, we can get started with our backing Store. Store is responsible for persiting the record data to Storage.

So remember that we have to validate the record bytes persisted using Store with the checksum and length? To make it easier to work with it, we create a virtual RecordHeader. This virtual RecordHeader is never actually persisted, but it is computed from the bytes to be written or bytes that are read from the storage.

#[derive(Debug, Default, Clone, Copy, PartialEq, Eq)]
pub struct RecordHeader {
    pub checksum: u64,
    pub length: u64,
}

In a previous index-less segmented-log implementation, RecordHeader instances used to be persisted right before every record on the Store. Once we moved record position, checksum and length metadata to the Index, it was no longer necessary to persist the RecordHeader.

We only need a constructor for RecordHeader:

impl RecordHeader {
    /// Computes the RecordHeader for the given serialized record bytes
    ///
    /// Returns a RecordHeader instance
    pub fn compute<H>(record_bytes: &[u8]) -> Self
    where
        H: Hasher + Default,
    {
        // compute the hash of the given record bytes
        let mut hasher = H::default();
        hasher.write(record_bytes);
        let checksum = hasher.finish();

        RecordHeader {
            checksum,
            length: record_bytes.len() as u64,
        }
    }
}

Now we can proceed with our Store implementation:

pub struct Store<S, H> {
    storage: S, /// Underlying storage

    /// Stores generic parameter used for Hasher impl.
    _phantom_data: PhantomData<H>,
}

impl<S, H> Store<S, H>
where
    S: Storage,
    H: Hasher + Default,
{
    /// Reads the bytes for the record persisted at the given position with
    /// the provided RecordHeader.
    ///
    /// This method reads and verifies the bytes read with the given
    /// RecordHeader by matching the checksum and length.
    ///
    /// Returns the bytes read for the record.
    pub async fn read(
        &self,
        position: &S::Position,
        record_header: &RecordHeader,
    ) -> Result<S::Content, StoreError<S::Error>> {
        // if this store is empty, error out
        if self.size() == u64_as_size!(0_u64, S::Size)? {
            return Err(StoreError::ReadOnEmptyStore);
        }

        // translate record_length usize to S::Size
        let record_length = record_header.length;
        let record_size = u64_as_size!(record_length, S::Size)?;

        // read the record bytes
        let record_bytes = self
            .storage
            .read(position, &record_size)
            .await
            .map_err(StoreError::StorageError)?;

        // cross verify the checksum and length of the bytes read
        if &RecordHeader::compute::<H>(&record_bytes) != record_header {
            return Err(StoreError::RecordHeaderMismatch);
        }

        Ok(record_bytes)
    }

    /// Appends the serialized stream of bytes slices for a record to this Store.
    ///
    /// Returns the computed RecordHeader for the bytes written.
    pub async fn append<XBuf, X, XE>(
        &mut self,
        stream: X,
        append_threshold: Option<S::Size>,
    ) -> Result<(S::Position, RecordHeader), StoreError<S::Error>>
    where
        XBuf: Deref<Target = [u8]>,
        X: Stream<Item = Result<XBuf, XE>> + Unpin,
    {
        let mut hasher = H::default();

        // compute a running checksum / hash
        let mut stream = stream.map(|x| match x {
            Ok(x) => {
                hasher.write(&x);
                Ok(x)
            }
            Err(e) => Err(e),
        });

        // append the byte slices to storage
        let (position, bytes_written) = self
            .storage
            .append(&mut stream, append_threshold)
            .await
            .map_err(StoreError::StorageError)?;

        // obtain the record header from the computed checksum and
        // length from bytes written
        let record_header = RecordHeader {
            checksum: hasher.finish(),
            length: size_as_u64!(bytes_written, S::Size)?,
        };

        Ok((position, record_header))
    }
}

We also need mechanism for constructing IndexRecord instances from RecordHeader instances once the record bytes are written to the store;

impl IndexRecord {
    /// Creates an IndexRecord from a store position and RecordHeader,
    /// presumably from a Store::append.
    pub fn with_position_and_record_header<P: ToPrimitive>(
        position: P,
        record_header: RecordHeader,
    ) -> Option<IndexRecord> {
        Some(IndexRecord {
            checksum: record_header.checksum,
            length: u32::try_from(record_header.length).ok()?,
            position: P::to_u32(&position)?,
        })
    }
}