Aerospike Software Engineer
Interview Questions & Guide 2026

import base64, json, os, struct, threading, zlib


class TinyLogKV:
    """One owning instance; logical index budget, not a process RAM cap."""
    MAX_FRAME = 8192

    def __init__(self, path, index_budget=130, hook=lambda phase: None):
        self.lock = threading.RLock()
        self.index, self.used, self.seq = {}, 0, 0
        self.budget, self.hook, self.poisoned = index_budget, hook, False
        self.file = open(path, 'r+b' if os.path.exists(path) else 'w+b', buffering=0)
        try:
            while True:
                offset = self.file.tell()
                record = self._read()
                if record is None:
                    break
                if record['seq'] != self.seq + 1:
                    raise ValueError('nonconsecutive sequence')
                self._apply(record, offset)
        except EOFError:
            try:
                self.file.truncate(offset)  # Only an incomplete final frame.
                self.file.flush(); os.fsync(self.file.fileno())
            except BaseException:
                self.file.close()
                raise
        except BaseException:
            self.file.close()
            raise
        self.file.seek(0, 2)

    def _read(self):
        header = self.file.read(4)
        if not header:
            return None
        if len(header) != 4:
            raise EOFError('incomplete header')
        size = struct.unpack('>I', header)[0]
        if not 1 <= size <= self.MAX_FRAME:
            raise ValueError('invalid frame size')
        body, crc = self.file.read(size), self.file.read(4)
        if len(body) != size or len(crc) != 4:
            raise EOFError('incomplete final frame')
        if zlib.crc32(body) != struct.unpack('>I', crc)[0]:
            raise ValueError('checksum mismatch')
        record = json.loads(body)
        if set(record) != {'seq', 'op', 'key', 'value'}:
            raise ValueError('invalid fields')
        if type(record['seq']) is not int or record['seq'] < 1:
            raise ValueError('invalid sequence')
        self._key(record['key'])
        if record['op'] not in ('put', 'delete'):
            raise ValueError('invalid operation')
        value = base64.b64decode(record['value'], validate=True)
        if len(value) > 1024 or (record['op'] == 'delete' and value):
            raise ValueError('invalid value')
        return record

    def _key(self, key):
        if not isinstance(key, str) or not 1 <= len(key.encode('utf8')) <= 64:
            raise ValueError('key must occupy 1..64 UTF-8 bytes')

    def _apply(self, record, offset):
        key = record['key']
        if key not in self.index:
            charge = 64 + len(key.encode('utf8'))
            if self.used + charge > self.budget:
                raise MemoryError('logical index budget exceeded')
            self.used += charge
        self.index[key] = (offset, record['seq'], record['op'] == 'delete')
        self.seq = record['seq']

    def _usable(self):
        if self.poisoned or self.file.closed:
            raise RuntimeError('close and reopen after an uncertain write')

    def get(self, key):
        self._key(key)
        with self.lock:
            self._usable()
            offset, version, deleted = self.index.get(key, (0, 0, True))
            if deleted:
                return None, version
            self.file.seek(offset)
            record = self._read()
            if record is None or record['key'] != key or record['seq'] != version:
                raise ValueError('index and log disagree')
            return base64.b64decode(record['value']), version

    def mutate(self, key, value, expected=None):
        self._key(key)
        if value is not None and (type(value) is not bytes or len(value) > 1024):
            raise ValueError('value must be bytes of length at most 1024, or None')
        if expected is not None and (type(expected) is not int or expected < 0):
            raise ValueError('expected version must be a nonnegative integer')
        with self.lock:
            self._usable()
            version = self.index.get(key, (0, 0, True))[1]
            if expected is not None and expected != version:
                return False
            charge = 0 if key in self.index else 64 + len(key.encode('utf8'))
            if self.used + charge > self.budget:
                raise MemoryError('logical index budget exceeded')
            record = {'seq': self.seq + 1, 'op': 'delete' if value is None else 'put',
                      'key': key, 'value': base64.b64encode(value or b'').decode()}
            body = json.dumps(record, separators=(',', ':')).encode()
            frame = struct.pack('>I', len(body)) + body + struct.pack('>I', zlib.crc32(body))
            offset = self.file.seek(0, 2)
            try:
                view = memoryview(frame)
                while view:
                    written = self.file.write(view)
                    if not written:
                        raise OSError('no write progress')
                    view = view[written:]
                self.file.flush(); os.fsync(self.file.fileno())
                self.hook('after_sync')
                self._apply(record, offset)
                self.hook('before_reply')
            except BaseException:
                self.poisoned = True
                raise
            return self.seq

    def close(self):
        with self.lock:
            self.file.close()

Scroll sideways to view long lines.