Object Storage

Object Storage (S3-like)

Object Storage -- система хранения неструктурированных данных (файлы, изображения, видео) с доступом по ключу. Проектируем аналог Amazon S3.

Шаг 1: Требования

Функциональные требования

1. CRUD операции над объектами (PUT, GET, DELETE, LIST)
2. Бакеты (buckets) для логической группировки
3. Поддержка объектов до 5 TB
4. Версионирование объектов
5. Метаданные (content-type, custom headers)
6. Multipart upload для больших файлов
7. Pre-signed URLs для временного доступа

Нефункциональные требования

1. Durability: 99.999999999% (11 nines)
2. Availability: 99.99%
3. Масштабирование до exabytes
4. Eventual consistency для LIST, strong consistency для GET after PUT
5. Низкая стоимость хранения

Шаг 2: High-Level архитектура

┌──────────┐     ┌───────────────┐     ┌──────────────────────────────────┐
│  Client  │────>│  API Gateway  │────>│  Object Storage Service          │
│          │     │  (S3 API)     │     │                                  │
└──────────┘     └───────────────┘     │  ┌────────────┐ ┌─────────────┐ │
                                       │  │ Metadata   │ │ Data        │ │
                                       │  │ Service    │ │ Service     │ │
                                       │  └─────┬──────┘ └──────┬──────┘ │
                                       └────────┼───────────────┼────────┘
                                                │               │
                                       ┌────────▼──────┐ ┌──────▼────────────┐
                                       │  Metadata DB  │ │  Data Nodes       │
                                       │  (PostgreSQL  │ │  ┌──────┐┌──────┐ │
                                       │   sharded)    │ │  │Node 1││Node 2│ │
                                       └───────────────┘ │  └──────┘└──────┘ │
                                                         │  ┌──────┐┌──────┐ │
                                                         │  │Node 3││Node N│ │
                                                         │  └──────┘└──────┘ │
                                                         └───────────────────┘

Шаг 3: Схема метаданных

CREATE TABLE buckets (
    id          UUID PRIMARY KEY DEFAULT gen_random_uuid(),
    name        VARCHAR(63) NOT NULL UNIQUE,
    owner_id    UUID        NOT NULL,
    region      VARCHAR(20) NOT NULL DEFAULT 'us-east-1',
    versioning  BOOLEAN     NOT NULL DEFAULT false,
    created_at  TIMESTAMPTZ NOT NULL DEFAULT now()
);

CREATE TABLE objects (
    id              UUID PRIMARY KEY DEFAULT gen_random_uuid(),
    bucket_id       UUID         NOT NULL REFERENCES buckets(id),
    key             TEXT         NOT NULL,
    version_id      UUID         NOT NULL DEFAULT gen_random_uuid(),
    size            BIGINT       NOT NULL,
    content_type    VARCHAR(255) NOT NULL DEFAULT 'application/octet-stream',
    etag            VARCHAR(64)  NOT NULL,
    metadata        JSONB        NOT NULL DEFAULT '{}',
    storage_class   VARCHAR(20)  NOT NULL DEFAULT 'STANDARD',
    is_deleted      BOOLEAN      NOT NULL DEFAULT false, -- for versioning
    chunks          JSONB        NOT NULL, -- chunk locations
    created_at      TIMESTAMPTZ  NOT NULL DEFAULT now(),
    UNIQUE (bucket_id, key, version_id)
);

CREATE INDEX idx_objects_bucket_key ON objects (bucket_id, key);
CREATE INDEX idx_objects_bucket_prefix ON objects (bucket_id, key text_pattern_ops);

CREATE TABLE chunks (
    id          UUID PRIMARY KEY DEFAULT gen_random_uuid(),
    object_id   UUID     NOT NULL REFERENCES objects(id),
    chunk_index INT      NOT NULL,
    size        BIGINT   NOT NULL,
    checksum    VARCHAR(64) NOT NULL,
    node_ids    UUID[]   NOT NULL, -- nodes where replicas are stored
    created_at  TIMESTAMPTZ NOT NULL DEFAULT now()
);

Шаг 4: Детальный дизайн

4.1 Upload Flow

<?php

declare(strict_types=1);

final class ObjectStorageService
{
    private const CHUNK_SIZE = 64 * 1024 * 1024; // 64 MB
    private const REPLICATION_FACTOR = 3;

    public function __construct(
        private readonly MetadataStore $metadata,
        private readonly DataNodeManager $dataNodes,
        private readonly ChunkPlacer $placer,
    ) {}

    /**
     * Upload an object (handles chunking automatically)
     */
    public function putObject(
        string $bucketName,
        string $key,
        StreamInterface $dataStream,
        array $metadata = [],
    ): ObjectInfo {
        $bucket = $this->metadata->getBucket($bucketName);

        // 1. Split data into chunks
        $chunks = [];
        $totalSize = 0;
        $hashCtx = hash_init('md5');
        $chunkIndex = 0;

        while (!$dataStream->eof()) {
            $chunkData = $dataStream->read(self::CHUNK_SIZE);
            $chunkSize = strlen($chunkData);

            if ($chunkSize === 0) {
                break;
            }

            hash_update($hashCtx, $chunkData);

            // 2. Select data nodes for this chunk (replica placement)
            $targetNodes = $this->placer->selectNodes(
                self::REPLICATION_FACTOR,
                $chunkSize,
            );

            // 3. Write chunk to data nodes
            $chunkChecksum = md5($chunkData);
            $chunkId = $this->writeChunkToNodes($chunkData, $targetNodes, $chunkChecksum);

            $chunks[] = [
                'id' => $chunkId,
                'index' => $chunkIndex,
                'size' => $chunkSize,
                'checksum' => $chunkChecksum,
                'nodes' => array_map(fn ($n) => $n->id, $targetNodes),
            ];

            $totalSize += $chunkSize;
            $chunkIndex++;
        }

        $etag = hash_final($hashCtx);

        // 4. Save metadata
        $objectId = $this->metadata->createObject(
            bucketId: $bucket->id,
            key: $key,
            size: $totalSize,
            etag: $etag,
            contentType: $metadata['content_type'] ?? 'application/octet-stream',
            metadata: $metadata,
            chunks: $chunks,
        );

        return new ObjectInfo(
            id: $objectId,
            key: $key,
            size: $totalSize,
            etag: $etag,
        );
    }

    /**
     * Download an object
     */
    public function getObject(
        string $bucketName,
        string $key,
        ?string $rangeHeader = null,
    ): ObjectDataStream {
        $object = $this->metadata->getObject($bucketName, $key);

        if ($object === null) {
            throw new ObjectNotFoundException("Object not found: {$key}");
        }

        // Parse range header for partial downloads
        $range = $rangeHeader ? $this->parseRange($rangeHeader, $object->size) : null;

        return new ObjectDataStream(
            object: $object,
            chunks: $object->chunks,
            dataNodes: $this->dataNodes,
            range: $range,
        );
    }

    /**
     * Delete an object
     */
    public function deleteObject(string $bucketName, string $key): void
    {
        $object = $this->metadata->getObject($bucketName, $key);

        if ($object === null) {
            return; // Idempotent
        }

        if ($object->bucket->versioning) {
            // Soft delete: add delete marker
            $this->metadata->addDeleteMarker($object->bucketId, $key);
        } else {
            // Hard delete: remove metadata and schedule chunk cleanup
            $this->metadata->deleteObject($object->id);
            $this->scheduleChunkCleanup($object->chunks);
        }
    }

    private function writeChunkToNodes(
        string $data,
        array $nodes,
        string $checksum,
    ): string {
        $chunkId = bin2hex(random_bytes(16));

        // Write to primary node, then replicate
        $primaryNode = $nodes[0];
        $primaryNode->writeChunk($chunkId, $data, $checksum);

        // Async replication to secondary nodes
        for ($i = 1; $i < count($nodes); $i++) {
            $this->asyncReplicate($chunkId, $data, $checksum, $nodes[$i]);
        }

        return $chunkId;
    }
}

4.2 Chunk Placement (Replica Distribution)

<?php

declare(strict_types=1);

final class ChunkPlacer
{
    public function __construct(
        private readonly DataNodeRegistry $registry,
    ) {}

    /**
     * Select nodes for chunk placement
     * Strategy: spread across different racks/AZs for durability
     */
    public function selectNodes(int $replicationFactor, int $chunkSize): array
    {
        $allNodes = $this->registry->getHealthyNodes();

        // Sort by available space (most space first)
        usort($allNodes, fn (DataNode $a, DataNode $b) =>
            $b->availableBytes <=> $a->availableBytes
        );

        // Select nodes from different racks
        $selected = [];
        $usedRacks = [];

        foreach ($allNodes as $node) {
            if (count($selected) >= $replicationFactor) {
                break;
            }

            // Prefer different racks for durability
            if (in_array($node->rackId, $usedRacks, true) && count($selected) < $replicationFactor - 1) {
                continue;
            }

            if ($node->availableBytes > $chunkSize) {
                $selected[] = $node;
                $usedRacks[] = $node->rackId;
            }
        }

        if (count($selected) < $replicationFactor) {
            throw new InsufficientStorageException(
                "Cannot find {$replicationFactor} nodes for chunk placement"
            );
        }

        return $selected;
    }
}

4.3 Multipart Upload

<?php

declare(strict_types=1);

final class MultipartUploadService
{
    public function __construct(
        private readonly MetadataStore $metadata,
        private readonly ObjectStorageService $storage,
        private readonly \Redis $redis,
    ) {}

    /**
     * Initiate multipart upload
     */
    public function initiate(string $bucketName, string $key, array $metadata = []): string
    {
        $uploadId = bin2hex(random_bytes(16));

        $this->redis->setex(
            "multipart:{$uploadId}",
            86400 * 7, // 7 days expiry
            json_encode([
                'bucket' => $bucketName,
                'key' => $key,
                'metadata' => $metadata,
                'parts' => [],
                'initiated_at' => time(),
            ]),
        );

        return $uploadId;
    }

    /**
     * Upload a part
     */
    public function uploadPart(
        string $uploadId,
        int $partNumber,
        string $data,
    ): PartInfo {
        $upload = $this->getUpload($uploadId);

        $etag = md5($data);

        // Store part temporarily
        $partKey = "multipart:{$uploadId}:part:{$partNumber}";
        $this->redis->set($partKey, $data);

        // Update parts list
        $upload['parts'][$partNumber] = [
            'part_number' => $partNumber,
            'size' => strlen($data),
            'etag' => $etag,
        ];

        $this->redis->setex(
            "multipart:{$uploadId}",
            86400 * 7,
            json_encode($upload),
        );

        return new PartInfo(
            partNumber: $partNumber,
            etag: $etag,
            size: strlen($data),
        );
    }

    /**
     * Complete multipart upload -- assemble all parts
     */
    public function complete(string $uploadId, array $partOrder): ObjectInfo
    {
        $upload = $this->getUpload($uploadId);

        // Validate all parts exist
        foreach ($partOrder as $partNumber) {
            if (!isset($upload['parts'][$partNumber])) {
                throw new \InvalidArgumentException("Part {$partNumber} not found");
            }
        }

        // Assemble stream from parts
        $assembledStream = new ConcatenatedStream();
        foreach ($partOrder as $partNumber) {
            $partData = $this->redis->get("multipart:{$uploadId}:part:{$partNumber}");
            $assembledStream->addPart($partData);
        }

        // Upload assembled object
        $result = $this->storage->putObject(
            $upload['bucket'],
            $upload['key'],
            $assembledStream,
            $upload['metadata'],
        );

        // Cleanup temporary parts
        $this->cleanupUpload($uploadId, $upload);

        return $result;
    }

    public function abort(string $uploadId): void
    {
        $upload = $this->getUpload($uploadId);
        $this->cleanupUpload($uploadId, $upload);
    }

    private function getUpload(string $uploadId): array
    {
        $data = $this->redis->get("multipart:{$uploadId}");
        if ($data === false) {
            throw new UploadNotFoundException("Upload not found: {$uploadId}");
        }
        return json_decode($data, true);
    }

    private function cleanupUpload(string $uploadId, array $upload): void
    {
        foreach (array_keys($upload['parts']) as $partNumber) {
            $this->redis->del("multipart:{$uploadId}:part:{$partNumber}");
        }
        $this->redis->del("multipart:{$uploadId}");
    }
}

4.4 Pre-signed URLs

<?php

declare(strict_types=1);

final class PreSignedUrlService
{
    public function __construct(
        private readonly string $secretKey,
        private readonly string $baseUrl,
    ) {}

    public function generateGetUrl(
        string $bucket,
        string $key,
        int $expiresIn = 3600,
    ): string {
        $expiry = time() + $expiresIn;

        $stringToSign = implode("\n", [
            'GET',
            $bucket,
            $key,
            $expiry,
        ]);

        $signature = hash_hmac('sha256', $stringToSign, $this->secretKey);

        return sprintf(
            '%s/%s/%s?expires=%d&signature=%s',
            $this->baseUrl,
            urlencode($bucket),
            urlencode($key),
            $expiry,
            $signature,
        );
    }

    public function validateSignature(
        string $method,
        string $bucket,
        string $key,
        int $expires,
        string $signature,
    ): bool {
        // Check expiration
        if (time() > $expires) {
            return false;
        }

        $stringToSign = implode("\n", [$method, $bucket, $key, $expires]);
        $expected = hash_hmac('sha256', $stringToSign, $this->secretKey);

        return hash_equals($expected, $signature);
    }
}

Шаг 5: Durability и Erasure Coding

Storage Classes

Возможные вопросы интервьюера

1. Как достичь 11 nines durability?

   • 3x replication across AZs
   • Erasure coding для экономии
   • Continuous data scrubbing (проверка checksums)
   • Автоматический repair при потере реплики

2. Как обрабатывать объекты размером 5 TB?

   • Multipart upload (части по 64 MB)
   • Параллельная загрузка частей
   • Resumable uploads при обрыве

3. Strong consistency vs Eventual?

   • GET after PUT: strong (читаем то, что записали)
   • LIST after PUT: eventually consistent (индекс обновляется асинхронно)

4. Как работает garbage collection?

   • Reference counting для chunks
   • Периодический background GC
   • Отложенное удаление (grace period 24h)