Design a disk-based cache for the client (platform independent, 32-byte keys, byte-array values, persistent, 100k+ objects, configurable max size 10MB-1GB, opaque and secure).

package com.androidinterview.diskcache;

import java.nio.ByteBuffer;
import java.util.Iterator;
import java.util.LinkedHashMap;
import java.util.Map;

// Every key that exists, in memory, and nothing else is. A hundred thousand
// entries at roughly 180 bytes of bookkeeping each, the key array, its
// ByteBuffer wrapper, the Location and the map entry, is about 18MB. That
// spend is what makes a read one positional read instead of a lookup on disk
// followed by a read, and it is still worth it.
//
// Keys are wrapped rather than used raw, because a byte array in a map is
// compared by identity and would never hit. Every method is synchronized,
// because a get on an access ordered map is a structural mutation, and two
// unguarded readers can corrupt it.
public final class CacheIndex {

    public record Location(int shard, long offset, int length) {

        // What the record costs on disk, header included. Counting only the
        // value would let 100k entries carry 4.8MB of headers that a 10MB
        // budget never saw.
        long recordBytes() {
            return Shard.HEADER_BYTES + length;
        }
    }

    // Access order, so the first key the iterator hands back is the least
    // recently used one and eviction never sorts.
    private final LinkedHashMap<ByteBuffer, Location> entries = new LinkedHashMap<>(16, 0.75f, true);
    private long liveBytes;

    public synchronized Location get(byte[] key) {
        return entries.get(ByteBuffer.wrap(key));
    }

    public synchronized void put(byte[] key, Location location) {
        Location previous = entries.put(ByteBuffer.wrap(key), location);
        if (previous != null) {
            liveBytes -= previous.recordBytes();
        }
        liveBytes += location.recordBytes();
    }

    public synchronized void remove(byte[] key) {
        Location removed = entries.remove(ByteBuffer.wrap(key));
        if (removed != null) {
            liveBytes -= removed.recordBytes();
        }
    }

    public synchronized byte[] leastRecentlyUsed() {
        Iterator<Map.Entry<ByteBuffer, Location>> oldestFirst = entries.entrySet().iterator();
        return oldestFirst.hasNext() ? oldestFirst.next().getKey().array() : null;
    }

    // Live bytes, not file bytes. The two diverge because overwritten and
    // evicted records stay on disk as holes until a compaction pass rewrites
    // the shard, and the size cap the caller configured is about live data.
    public synchronized long liveBytes() {
        return liveBytes;
    }
}

package com.androidinterview.diskcache;

import java.io.Closeable;
import java.io.File;
import java.io.IOException;
import java.nio.ByteBuffer;

// The cache itself. Opaque means it never looks inside a value, so the whole
// public surface is two byte arrays in and one byte array out.
public final class DiskCache implements Closeable {

    private static final byte[] TOMBSTONE = new byte[0];

    private final Shard[] shards;
    private final CacheIndex index = new CacheIndex();
    private final ValueCipher cipher;
    private final long maxBytes;
    // Writes serialize here, so the order of records in the logs always
    // matches the order the index saw them, which is what replay trusts.
    // Reads never take it. The expensive half of a write, the encryption,
    // happens before the lock is taken.
    private final Object writeLock = new Object();

    public DiskCache(File directory, int shardCount, long maxBytes, ValueCipher cipher) throws IOException {
        this.shards = new Shard[shardCount];
        this.cipher = cipher;
        this.maxBytes = maxBytes;
        directory.mkdirs();
        for (int i = 0; i < shardCount; i++) {
            shards[i] = new Shard(new File(directory, "shard-" + i + ".log"));
            final int shard = i;
            // Replay is in file order, which is write order, so a later record
            // for a key overwrites the earlier one exactly as it did at write
            // time. Recency ordering restarts as insertion order, which is an
            // approximation the first few reads correct.
            shards[i].replay((key, timestamp, offset, length) -> {
                if (length == 0) {
                    index.remove(key);
                } else {
                    index.put(key, new CacheIndex.Location(shard, offset, length));
                }
            });
        }
    }

    // Look up offset and length in memory, one positional read, then open the
    // sealed bytes. A miss costs nothing, because the index is complete and
    // never has to consult a file to know an entry is absent.
    public byte[] get(byte[] key) throws IOException {
        CacheIndex.Location at = index.get(key);
        if (at == null) {
            return null;
        }
        byte[] value = cipher.open(key, shards[at.shard()].read(at.offset(), at.length()));
        if (value == null) {
            // The tag did not verify, so the record is damaged or tampered.
            // Evict it now, or the key stays poisoned in the index and every
            // later read pays for the same failed decrypt.
            tombstone(key);
        }
        return value;
    }

    public void put(byte[] key, byte[] value) throws IOException {
        byte[] sealed = cipher.seal(key, value);
        synchronized (writeLock) {
            int shard = shardFor(key);
            long offset = shards[shard].append(key, sealed, System.currentTimeMillis());
            index.put(key, new CacheIndex.Location(shard, offset, sealed.length));
            trim();
        }
    }

    public void remove(byte[] key) throws IOException {
        tombstone(key);
    }

    // True once dead bytes on disk outweigh live ones. This class decides
    // when compaction is worth it and the caller decides where it runs, a
    // background job that rewrites a shard's live records into a fresh file
    // and swaps it in. Without that pass a write heavy workload grows the
    // files past the configured cap indefinitely.
    public boolean compactionDue() throws IOException {
        long fileBytes = 0;
        for (Shard shard : shards) {
            fileBytes += shard.fileBytes();
        }
        return fileBytes - index.liveBytes() > index.liveBytes();
    }

    // Evict until the live total is back under the configured cap. The space
    // is not reclaimed here, only the entry, and compaction reclaims it later.
    private void trim() throws IOException {
        while (index.liveBytes() > maxBytes) {
            byte[] victim = index.leastRecentlyUsed();
            if (victim == null) {
                return;
            }
            tombstone(victim);
        }
    }

    // A delete in an append only log is itself a record, a zero length one.
    // Without the tombstone the next replay would find the original write and
    // bring the entry back from the dead.
    private void tombstone(byte[] key) throws IOException {
        synchronized (writeLock) {
            shards[shardFor(key)].append(key, TOMBSTONE, System.currentTimeMillis());
            index.remove(key);
        }
    }

    // The key is already an opaque 32 byte hash handed to us by the caller, so
    // its bits are as uniform as anything we could compute from it. Hashing a
    // hash buys nothing, and four bytes of it cover any sane shard count.
    private int shardFor(byte[] key) {
        return Math.floorMod(ByteBuffer.wrap(key).getInt(), shards.length);
    }

    @Override
    public void close() throws IOException {
        for (Shard shard : shards) {
            shard.close();
        }
    }
}

package com.androidinterview.diskcache;

import java.io.Closeable;
import java.io.EOFException;
import java.io.File;
import java.io.IOException;
import java.nio.ByteBuffer;
import java.nio.channels.FileChannel;
import java.nio.file.StandardOpenOption;
import java.util.zip.CRC32;

// One append only log file. Sharding by key bounds how large any single file
// gets and keeps a compaction pass small. Appends take the shard's own lock,
// and reads take no lock at all, because a positional read on a channel is
// seek free and safe from any number of threads at once.
public final class Shard implements Closeable {

    public static final int KEY_BYTES = 32;

    // A 32 byte key, a four byte length, an eight byte timestamp and a four
    // byte checksum over the three of them, all fixed width and big endian.
    // Nothing here is Java serialization, because an iOS or a desktop build of
    // the same spec has to read these files. The timestamp is unused by the
    // pure LRU cache and is there for the TTL variant the clarifying question
    // asks about, so the format does not change when that lands.
    public static final int HEADER_BYTES = KEY_BYTES + 4 + 8 + 4;
    private static final int CHECKED_BYTES = HEADER_BYTES - 4;

    public interface RecordVisitor {
        void visit(byte[] key, long timestamp, long offset, int length);
    }

    private final FileChannel channel;

    public Shard(File path) throws IOException {
        // Opening with DSYNC as well would survive power loss at the price of
        // a flush per append. As written, an append survives the process being
        // killed, which is the failure a cache actually meets.
        this.channel = FileChannel.open(
                path.toPath(), StandardOpenOption.CREATE, StandardOpenOption.READ, StandardOpenOption.WRITE);
    }

    // Writes only ever append. A process killed mid write damages the last
    // record and nothing before it, which turns crash recovery into a
    // truncation rather than a repair.
    public synchronized long append(byte[] key, byte[] sealed, long timestamp) throws IOException {
        long offset = channel.size();
        ByteBuffer record = ByteBuffer.allocate(HEADER_BYTES + sealed.length);
        record.put(key).putInt(sealed.length).putLong(timestamp);
        record.putInt(headerChecksum(record.array()));
        record.put(sealed).flip();
        long position = offset;
        while (record.hasRemaining()) {
            position += channel.write(record, position);
        }
        return offset;
    }

    // One positional read, because the caller already knows the offset and
    // the length. There is no index on disk to consult first, and no seek to
    // race another reader on.
    public byte[] read(long offset, int length) throws IOException {
        ByteBuffer sealed = ByteBuffer.allocate(length);
        readFully(sealed, offset + HEADER_BYTES);
        return sealed.array();
    }

    // Startup replays each shard once to rebuild the index. Replay stops at
    // the first header whose checksum does not match or whose length runs
    // past the end of the file, because that is the tail of a write that
    // never finished, and truncates the file back to the last whole record. A
    // damaged value body is not checked here, the cipher's tag catches that on
    // first read and the cache evicts the entry then.
    public void replay(RecordVisitor visitor) throws IOException {
        long end = channel.size();
        long offset = 0;
        ByteBuffer header = ByteBuffer.allocate(HEADER_BYTES);
        while (offset + HEADER_BYTES <= end) {
            header.clear();
            readFully(header, offset);
            header.flip();
            byte[] key = new byte[KEY_BYTES];
            header.get(key);
            int length = header.getInt();
            long timestamp = header.getLong();
            int checksum = header.getInt();
            if (checksum != headerChecksum(header.array()) || length < 0 || offset + HEADER_BYTES + length > end) {
                break;
            }
            visitor.visit(key, timestamp, offset, length);
            offset += HEADER_BYTES + length;
        }
        channel.truncate(offset);
    }

    public long fileBytes() throws IOException {
        return channel.size();
    }

    private void readFully(ByteBuffer into, long position) throws IOException {
        while (into.hasRemaining()) {
            int read = channel.read(into, position);
            if (read < 0) {
                throw new EOFException("Record runs past the end of the shard");
            }
            position += read;
        }
    }

    private static int headerChecksum(byte[] header) {
        CRC32 crc = new CRC32();
        crc.update(header, 0, CHECKED_BYTES);
        return (int) crc.getValue();
    }

    @Override
    public void close() throws IOException {
        channel.close();
    }
}

package com.androidinterview.diskcache;

// Secure is a port, not an algorithm this cache gets to pick. The
// implementation is AES GCM per value, with the key generated in and held by
// the platform keystore, so the key material never sits beside the data it
// protects. The byte layout is portable, the key is not, so a cross platform
// build shares the spec and never the ciphertext.
//
// Per value rather than per file, because decrypting one entry must not mean
// decrypting a shard. The sealed bytes are a fresh twelve byte random nonce,
// then the ciphertext, then GCM's sixteen byte tag. The nonce is generated
// per seal and never reused, because a repeated nonce under one key is the
// one way to break GCM. The tag is the tamper check, which is why the value
// needs no checksum of its own.
public interface ValueCipher {

    byte[] seal(byte[] key, byte[] value);

    // Returns null when the tag does not verify, which covers both a tampered
    // record and bit rot in a value body that replay does not inspect.
    byte[] open(byte[] key, byte[] sealed);
}

package com.androidinterview.diskcache

import java.nio.ByteBuffer

// Every key that exists, in memory, and nothing else is. A hundred thousand
// entries at roughly 180 bytes of bookkeeping each, the key array, its
// ByteBuffer wrapper, the Location and the map entry, is about 18MB. That
// spend is what makes a read one positional read rather than a lookup on disk
// followed by a read, and it is still worth it.
//
// Keys are wrapped, because a ByteArray in a map is compared by identity and
// would never hit. Access order means the first key the iterator hands back is
// the least recently used one, so eviction never sorts. Every method is
// synchronized, because a get on an access ordered map is a structural
// mutation, and two unguarded readers can corrupt it.
class CacheIndex {

    data class Location(val shard: Int, val offset: Long, val length: Int) {
        // What the record costs on disk, header included. Counting only the
        // value would let 100k entries carry 4.8MB of headers that a 10MB
        // budget never saw.
        val recordBytes get() = Shard.HEADER_BYTES + length.toLong()
    }

    private val entries = LinkedHashMap<ByteBuffer, Location>(16, 0.75f, true)

    // Live bytes, not file bytes. The two diverge because overwritten and
    // evicted records stay on disk as holes until compaction rewrites the
    // shard, and the cap the caller configured is about live data.
    var liveBytes = 0L
        @Synchronized get
        private set

    @Synchronized
    operator fun get(key: ByteArray): Location? = entries[ByteBuffer.wrap(key)]

    @Synchronized
    operator fun set(key: ByteArray, location: Location) {
        entries.put(ByteBuffer.wrap(key), location)?.let { liveBytes -= it.recordBytes }
        liveBytes += location.recordBytes
    }

    @Synchronized
    fun remove(key: ByteArray) {
        entries.remove(ByteBuffer.wrap(key))?.let { liveBytes -= it.recordBytes }
    }

    @Synchronized
    fun leastRecentlyUsed(): ByteArray? = entries.keys.firstOrNull()?.array()
}

package com.androidinterview.diskcache

import java.io.Closeable
import java.io.File
import java.nio.ByteBuffer

// The cache itself. Opaque means it never looks inside a value, so the whole
// public surface is two byte arrays in and one byte array out.
class DiskCache(
    directory: File,
    shardCount: Int,
    private val maxBytes: Long,
    private val cipher: ValueCipher,
) : Closeable {

    private val index = CacheIndex()

    // Writes serialize here, so the order of records in the logs always
    // matches the order the index saw them, which is what replay trusts.
    // Reads never take it. The expensive half of a write, the encryption,
    // happens before the lock is taken.
    private val writeLock = Any()

    private val shards = directory.mkdirs().let {
        List(shardCount) { i ->
            Shard(File(directory, "shard-$i.log")).also { shard ->
                // Replay is in file order, which is write order, so a later
                // record for a key overwrites the earlier one exactly as it did
                // at write time. Recency restarts as insertion order, an
                // approximation the first few reads correct.
                shard.replay { key, _, offset, length ->
                    if (length == 0) index.remove(key) else index[key] = CacheIndex.Location(i, offset, length)
                }
            }
        }
    }

    // Look up offset and length in memory, one positional read, then open the
    // sealed bytes. A miss costs nothing, because the index is complete and
    // never consults a file to know an entry is absent.
    operator fun get(key: ByteArray): ByteArray? {
        val at = index[key] ?: return null
        val value = cipher.open(key, shards[at.shard].read(at.offset, at.length))
        // A null from the cipher means the tag did not verify, so the record
        // is damaged or tampered. Evict it now, or the key stays poisoned in
        // the index and every later read pays for the same failed decrypt.
        if (value == null) tombstone(key)
        return value
    }

    fun put(key: ByteArray, value: ByteArray) {
        val sealed = cipher.seal(key, value)
        synchronized(writeLock) {
            val shard = shardFor(key)
            index[key] = CacheIndex.Location(shard, shards[shard].append(key, sealed, now()), sealed.size)
            trim()
        }
    }

    fun remove(key: ByteArray) = tombstone(key)

    // True once dead bytes on disk outweigh live ones. This class decides when
    // compaction is worth it and the caller decides where it runs, a
    // background job that rewrites a shard's live records into a fresh file
    // and swaps it in. Without that pass a write heavy workload grows the
    // files past the configured cap indefinitely.
    fun compactionDue(): Boolean = shards.sumOf(Shard::fileBytes) - index.liveBytes > index.liveBytes

    // Evict until the live total is back under the configured cap. The space
    // is not reclaimed here, only the entry, and compaction reclaims it later.
    private fun trim() {
        while (index.liveBytes > maxBytes) tombstone(index.leastRecentlyUsed() ?: return)
    }

    // A delete in an append only log is itself a record, a zero length one.
    // Without the tombstone the next replay would find the original write and
    // bring the entry back from the dead.
    private fun tombstone(key: ByteArray) = synchronized(writeLock) {
        shards[shardFor(key)].append(key, ByteArray(0), now())
        index.remove(key)
    }

    // The key is already an opaque 32 byte hash handed to us by the caller, so
    // its bits are as uniform as anything we could compute from it. Hashing a
    // hash buys nothing, and four bytes of it cover any sane shard count.
    private fun shardFor(key: ByteArray) = Math.floorMod(ByteBuffer.wrap(key).int, shards.size)

    private fun now() = System.currentTimeMillis()

    override fun close() = shards.forEach(Shard::close)
}

package com.androidinterview.diskcache

import java.io.Closeable
import java.io.EOFException
import java.io.File
import java.nio.ByteBuffer
import java.nio.channels.FileChannel
import java.nio.file.StandardOpenOption.CREATE
import java.nio.file.StandardOpenOption.READ
import java.nio.file.StandardOpenOption.WRITE
import java.util.zip.CRC32

// One append only log file. Sharding by key bounds how large any single file
// gets and keeps a compaction pass small. Appends take the shard's own lock,
// and reads take no lock at all, because a positional read on a channel is
// seek free and safe from any number of threads at once.
class Shard(path: File) : Closeable {

    // Opening with DSYNC as well would survive power loss at the price of a
    // flush per append. As written, an append survives the process being
    // killed, which is the failure a cache actually meets.
    private val channel = FileChannel.open(path.toPath(), CREATE, READ, WRITE)

    // Writes only ever append. A process killed mid write damages the last
    // record and nothing before it, which turns crash recovery into a
    // truncation rather than a repair.
    @Synchronized
    fun append(key: ByteArray, sealed: ByteArray, timestamp: Long): Long {
        val offset = channel.size()
        val record = ByteBuffer.allocate(HEADER_BYTES + sealed.size)
        record.put(key).putInt(sealed.size).putLong(timestamp)
        record.putInt(headerChecksum(record.array()))
        record.put(sealed).flip()
        var position = offset
        while (record.hasRemaining()) position += channel.write(record, position)
        return offset
    }

    // One positional read, because the caller already knows the offset and
    // the length. There is no index on disk to consult first, and no seek to
    // race another reader on.
    fun read(offset: Long, length: Int): ByteArray =
        ByteBuffer.allocate(length).also { readFully(it, offset + HEADER_BYTES) }.array()

    // Startup replays each shard once to rebuild the index. Replay stops at
    // the first header whose checksum does not match or whose length runs
    // past the end of the file, because that is the tail of a write that
    // never finished, and truncates the file back to the last whole record. A
    // damaged value body is not checked here, the cipher's tag catches that on
    // first read and the cache evicts the entry then.
    fun replay(visit: (key: ByteArray, timestamp: Long, offset: Long, length: Int) -> Unit) {
        val end = channel.size()
        var offset = 0L
        val header = ByteBuffer.allocate(HEADER_BYTES)
        while (offset + HEADER_BYTES <= end) {
            header.clear()
            readFully(header, offset)
            header.flip()
            val key = ByteArray(KEY_BYTES).also(header::get)
            val length = header.int
            val timestamp = header.long
            val checksum = header.int
            if (checksum != headerChecksum(header.array()) || length < 0 || offset + HEADER_BYTES + length > end) break
            visit(key, timestamp, offset, length)
            offset += HEADER_BYTES + length
        }
        channel.truncate(offset)
    }

    fun fileBytes(): Long = channel.size()

    private fun readFully(into: ByteBuffer, from: Long) {
        var position = from
        while (into.hasRemaining()) {
            val read = channel.read(into, position)
            if (read < 0) throw EOFException("Record runs past the end of the shard")
            position += read
        }
    }

    override fun close() = channel.close()

    companion object {
        const val KEY_BYTES = 32

        // A 32 byte key, a four byte length, an eight byte timestamp and a
        // four byte checksum over the three of them, all fixed width and big
        // endian. Nothing here is Java serialization, because an iOS or a
        // desktop build of the same spec reads these files. The timestamp is
        // unused by the pure LRU cache and is there for the TTL variant the
        // clarifying question asks about, so the format does not change when
        // that lands.
        const val HEADER_BYTES = KEY_BYTES + 4 + 8 + 4
        private const val CHECKED_BYTES = HEADER_BYTES - 4

        private fun headerChecksum(header: ByteArray): Int =
            CRC32().apply { update(header, 0, CHECKED_BYTES) }.value.toInt()
    }
}

package com.androidinterview.diskcache

// Secure is a port, not an algorithm this cache gets to pick. The
// implementation is AES GCM per value, with the key generated in and held by
// the platform keystore, so the key material never sits beside the data it
// protects. The byte layout is portable, the key is not, so a cross platform
// build shares the spec and never the ciphertext.
//
// Per value rather than per file, because decrypting one entry must not mean
// decrypting a shard. The sealed bytes are a fresh twelve byte random nonce,
// then the ciphertext, then GCM's sixteen byte tag. The nonce is generated
// per seal and never reused, because a repeated nonce under one key is the
// one way to break GCM. The tag is the tamper check, which is why the value
// needs no checksum of its own.
interface ValueCipher {

    fun seal(key: ByteArray, value: ByteArray): ByteArray

    // Null when the tag does not verify, which covers both a tampered record
    // and bit rot in a value body that replay does not inspect.
    fun open(key: ByteArray, sealed: ByteArray): ByteArray?
}