Distributed Systems

34 posts

woowahanOriginal article

아한형제들 기술블로그 (opens in new tab)

The 7th Woowacourse crew has successfully launched three distinct services, demonstrating that modern software engineering requires a synergy of technical mastery and "soft skills" like product planning and team communication. By owning the entire lifecycle from ideation to deployment, these developers moved beyond mere coding to solve real-world problems through agile iterations, user feedback, and robust infrastructure management. The program’s focus on the full stack of development—including monitoring, 2-week sprints, and collaborative design—highlights a shift toward producing well-rounded engineers capable of navigating professional environments. ### The Woowacourse Full-Cycle Philosophy * The 10-month curriculum emphasizes soft skills, including speaking and writing, alongside traditional technical tracks like Web Backend, Frontend, and Mobile Android. * During Level 3 and 4, crews transition from fundamental programming to managing team projects where they must handle everything from initial architecture to UI/UX design. * The process mimics real-world industry standards by implementing 2-week development sprints, establishing monitoring environments, and managing automated deployment pipelines. * The core goal is to shift the developer's mindset from simply writing code to understanding why certain features are planned and how architecture choices impact the final user value. ### Pickeat: Collaborative Dining Decisions * This service addresses "decision fatigue" during group meals by providing a collaborative platform to filter restaurants based on dietary constraints and preferences. * Technical challenges included frequent domain restructuring and UI overhauls as the team pivoted based on real-world user feedback during demo days. * The platform utilizes location data for automatic restaurant lookups and supports real-time voting mechanisms to ensure democratic and efficient group decisions. * Development focused on aligning team judgment standards and iterating quickly to validate product-market fit rather than adhering strictly to initial specifications. ### Bottari: Real-Time Synchronized Checklists * Bottari is a checklist service designed for situations like traveling or moving, focusing on "becoming a companion for the user’s memory." * The service features template-based list generation and a "Team Bottari" function that allows multiple users to collaborate on a single list with real-time synchronization. * A major technical focus was placed on the user experience flow, specifically optimizing notification timing and sync states to provide "peace of mind" for users. * The project demonstrates the principle that technology serves as a tool for solving psychological pain points, such as the anxiety of forgetting essential items. ### Coffee Shout: Real-Time Betting and Mini-Games * Designed to gamify office culture, this service replaces simple "rock-paper-scissors" with interactive mini-games and weighted roulette for coffee bets. * The technical stack involved challenging implementations of WebSockets and distributed environments to handle the concurrency required for real-time gaming. * The team focused on algorithm balancing for the weighted roulette system to ensure fairness and excitement during the betting process. * Refinement of the service was driven by direct feedback from other Woowacourse crews, emphasizing the importance of community testing in the development lifecycle. These projects underscore that the transition from a student to a professional developer is defined by the ability to manage shifting requirements and technical complexity while maintaining a focus on the end-user's experience.

dropbox3 min readCurated summary

Building the future: highlights from Dropbox’s 2025 summer intern class

Dropbox’s 2025 intern program brought together 43 interns from 27 universities for 12 weeks of mentorship, technical work, and community-building. The 28 engineering interns contributed to systems spanning AI, search, storage, data infrastructure, and developer tools, with many projects supporting Dropbox Dash. Their work demonstrates how targeted refactoring, automation, and intelligent infrastructure can improve reliability, reduce costs, and expand product capabilities. ## A Diverse, Mentorship-Focused Internship Program - Interns received more than 6,000 hours of one-on-one mentorship. - The cohort included students from institutions in the United States, Canada, Poland, and Ireland. - Programming included Virtual First events, employee resource group activities, and an in-person Emerging Talent Summit. - Projects were aligned with Dropbox’s production systems and company goals rather than being isolated experiments. ## Infrastructure and Reliability Improvements - **Filesystem Data:** Rhea Rai redesigned Dropbox’s file history tracking system, emphasizing strongly tested code and simpler metadata infrastructure while reducing operational costs. - **Storage Core:** Albert Joon Sung reduced Magic Pocket write latency during disk restarts by adding storage-health caching and routing writes away from degraded volumes. - **Metrics:** Yonatan Ginsburg developed adaptive anomaly detection for Vortex2, accounting for changing patterns and seasonality to reduce alert fatigue and improve incident response. - **Analytics Platform:** Sanjith Udupa built recommendations for optimizing Databricks queries and ETL pipelines, and documented a plan to migrate a 500 TB mobile-events dataset to liquid clustering. ## AI, Search, and Dropbox Dash - **ML Platform:** Ben Juntilla created AI Sentinel, which gives engineers real-time visibility into machine-learning deployment health and improves confidence in model releases. - **Connector Platform:** Eddie Ormseth built tools that provide access to fresher Dash persistence data and additional third-party metadata without requiring connector teams to reprocess data. - **Retrieval Platform:** Rishi Peddakama expanded unified search to more than 20 languages by integrating language detection into indexing and retrieval. - **Find & Discover:** Francesca Venditti created in-context document previews for Dash, including PDF viewing and links to AI-powered follow-up chat. - **Conversational AI:** Alan Zhu developed a modular web-automation agent and connected it to Dropbox APIs for actions such as searching for and uploading files. ## Developer Automation - Ahmed Ibrahim built an AI-assisted code migration tool on Dropbox’s internal migration platform. - Developers can run migrations on selected folders, configure them through a CLI or automated workflow, and receive pull requests automatically when jobs succeed. - The tool enabled two major migrations during the internship and illustrates how automation can reduce repetitive engineering work. ## Broader Impact The interns’ projects improved system performance, operational visibility, multilingual accessibility, data freshness, and developer productivity. Together, they supported Dropbox’s move toward AI-first products while reinforcing the company’s emphasis on trustworthy, maintainable, and efficient engineering. For students interested in production-scale software, AI, and infrastructure, Dropbox presents its internship program and open roles as opportunities to contribute directly to products such as Dropbox Dash.

Read original(opens in new tab)
naverOriginal article

Naver TV (opens in new tab)

The Apache Kafka ecosystem is undergoing a significant architectural shift with the introduction of Consumer Group Protocol v2, as outlined in KIP-848. This update addresses long-standing performance bottlenecks and stability issues inherent in the original client-side rebalancing logic by moving the responsibility of partition assignment to the broker. This change effectively eliminates the "stop-the-world" effect during rebalances and significantly improves the scalability of large-scale consumer groups. ### Limitations of the Legacy Consumer Group Protocol (v1) * **Heavy Client-Side Logic:** In v1, the "Group Leader" (a specific consumer instance) is responsible for calculating partition assignments, which creates a heavy burden on the client and leads to inconsistent behavior across different programming language implementations. * **Stop-the-World Rebalancing:** Whenever a member joins or leaves the group, all consumers must stop processing data until the new assignment is synchronized, leading to significant latency spikes. * **Sensitivity to Processing Delays:** Because heartbeats and data processing often share the same thread, a slow consumer can trigger a session timeout, causing an unnecessary and disruptive group rebalance. ### Architectural Improvements in Protocol v2 * **Server-Side Reconciliation:** The reconciliation logic is moved to the Group Coordinator on the broker, simplifying the client and ensuring that partition assignment is managed centrally and consistently. * **Incremental Rebalancing:** Unlike the "eager" rebalancing of v1, the new protocol allows consumers to keep their existing partitions while negotiating new ones, ensuring continuous data processing. * **Decoupled Heartbeats:** The heartbeat mechanism is separated from the main processing loop, preventing "zombie member" scenarios where a busy consumer is incorrectly marked as dead. ### Performance and Scalability Gains * **Reduced Rebalance Latency:** By offloading the assignment logic to the broker, the time required to stabilize a group after a membership change is reduced from seconds to milliseconds. * **Large-Scale Group Support:** The new protocol is designed to handle thousands of partitions and hundreds of consumers within a single group without the exponential performance degradation seen in v1. * **Stable Deployments:** During rolling restarts or deployments, the group remains stable and avoids the "rebalance storms" that typically occur when multiple instances cycle at once. ### Migration and Practical Implementation * **Configuration Requirements:** Users can opt-in to the new protocol by setting the `group.protocol` configuration to `consumer` (introduced as early access in Kafka 3.7 and standard in 4.0). * **Compatibility:** While the new protocol requires updated brokers and clients, it is designed to support a transition phase to allow organizations to migrate their workloads gradually. * **New Tooling:** Updated command-line tools and metrics are provided to monitor the server-side assignment process and track group state more granularly. Organizations experiencing frequent rebalance issues or managing high-throughput Kafka clusters should plan for a migration to Consumer Group Protocol v2. Transitioning to this server-side assignment model is highly recommended for stabilizing production environments and reducing the operational overhead associated with consumer group management.

netflixOriginal article

How and Why Netflix Built a Real-Time Distributed Graph: Part 1 — Ingesting and Processing Data Streams at Internet Scale | by Netflix Technology Blog | Netflix TechBlog (opens in new tab)

Netflix has developed a Real-Time Distributed Graph (RDG) to unify member interaction data across its expanding business verticals, including streaming, live events, and mobile gaming. By transitioning from siloed microservice data to a graph-based model, the company can perform low-latency, relationship-centric queries that were previously hindered by expensive manual joins and data fragmentation. The resulting system enables Netflix to track user journeys across various devices and platforms in real-time, providing a foundation for deeper personalization and pattern detection. ### Challenges of Data Isolation in Microservices * While Netflix’s microservices architecture facilitates independent scaling and service decomposition, it inherently leads to data isolation where each service manages its own storage. * Data scientists and engineers previously had to "stitch" together disparate data from various databases and the central data warehouse, which was a slow and manual process. * The RDG moves away from table-based models to a relationship-centric model, allowing for efficient "hops" across nodes without the need for complex denormalization. * This flexibility allows the system to adapt to new business entities (like live sports or games) without requiring massive schema re-architectures. ### Real-Time Ingestion and Normalization * The ingestion layer is designed to capture events from diverse upstream sources, including Change Data Capture (CDC) from databases and request/response logs. * Netflix utilizes its internal data pipeline, Keystone, to funnel these high-volume event streams into the processing framework. * The system must handle "Internet scale" data, ensuring that events from millions of members are captured as they happen to maintain an up-to-date view of the graph. ### Stream Processing with Apache Flink * Netflix uses Apache Flink as the core stream processing engine to handle the transformation of raw events into graph entities. * Incoming data undergoes normalization to ensure a standardized format, regardless of which microservice or business vertical the data originated from. * The pipeline performs data enrichment, joining incoming streams with auxiliary metadata to provide a comprehensive context for each interaction. * The final step of the processing layer involves mapping these enriched events into a graph structure of nodes (entities) and edges (relationships), which are then emitted to the system's storage layer. ### Practical Conclusion Organizations operating with a highly decoupled microservices architecture should consider a graph-based ingestion strategy to overcome the limitations of data silos. By leveraging stream processing tools like Apache Flink to build a real-time graph, engineering teams can provide stakeholders with the ability to discover hidden relationships and cross-domain insights that are often lost in traditional data warehouses.

netflixOriginal article

Behind the Streams: Real-Time Recommendations for Live Events Part 3 | by Netflix Technology Blog | Netflix TechBlog (opens in new tab)

Netflix manages the massive surge of concurrent users during live events by utilizing a hybrid strategy of prefetching and real-time broadcasting to deliver synchronized recommendations. By decoupling data delivery from the live trigger, the system avoids the "thundering herd" effect that would otherwise overwhelm cloud infrastructure during record-breaking broadcasts. This architecture ensures that millions of global devices receive timely updates and visual cues without requiring linear, inefficient scaling of compute resources. ### The Constraint Optimization Problem To maintain a seamless experience, Netflix engineers balance three primary technical constraints: time to update, request throughput, and compute cardinality. * **Time:** The specific duration required to coordinate and push a recommendation update to the entire global fleet. * **Throughput:** The maximum capacity of cloud services to handle incoming requests without service degradation. * **Cardinality:** The variety and complexity of unique requests necessary to serve personalized updates to different user segments. ### Two-Phase Recommendation Delivery The system splits the delivery process into two distinct stages to smooth out traffic spikes and ensure high availability. * **Prefetching Phase:** While members browse the app normally before an event, the system downloads materialized recommendations, metadata, and artwork into the device's local cache. * **Broadcasting Phase:** When the event begins, a low-cardinality "at least once" message is broadcast to all connected devices, triggering them to display the already-cached content instantaneously. * **Traffic Smoothing:** This approach eliminates the need for massive, real-time data fetches at the moment of kickoff, distributing the heavy lifting of data transfer over a longer period. ### Live State Management and UI Synchronization A dedicated Live State Management (LSM) system tracks event schedules in real time to ensure the user interface stays perfectly in sync with the production. * **Dynamic Adjustments:** If a live event is delayed or ends early, the LSM adjusts the broadcast triggers to preserve accuracy and prevent "spoilers" or dead links. * **Visual Cues:** The UI utilizes "Live" badging and dynamic artwork transitions to signal urgency and guide users toward the stream. * **Frictionless Playback:** For members already on a title’s detail page, the system can trigger an automatic transition into the live player the moment the broadcast begins, reducing navigation latency. To support global-scale live events, technical teams should prioritize edge-heavy strategies that pre-position assets on client devices. By shifting from a reactive request-response model to a proactive prefetch-and-trigger model, platforms can maintain high performance and reliability even during the most significant traffic peaks.

airbnb3 min readCurated summary

From Static Rate Limiting to Adaptive Traffic Management in Airbnb’s Key-Value Store

Airbnb evolved Mussel’s QoS system from static, per-client QPS limits into adaptive traffic management designed to maximize goodput. The newer approach accounts for the actual cost of requests, prioritizes critical workloads under stress, and detects hot keys or attack traffic before they overwhelm storage. Together, resource-aware quotas and real-time load shedding provide stronger protection against traffic spikes, uneven workloads, and DDoS-like bursts. ## Why Static QPS Limits Fell Short - Mussel is a multi-tenant key-value store serving millions of point and range reads across Airbnb. - Its original Redis-backed limiter assigned each client a fixed requests-per-second quota. - Requests exceeding the quota received HTTP 429 responses. - This model worked when backend effort roughly matched request count. - As usage grew, it could not account for: - The difference between a cheap one-row lookup and a 100,000-row scan. - Hot keys accessed by many clients simultaneously. - Localized storage-shard overload that affected unrelated traffic. - Sudden events such as bot floods, DDoS attacks, or large uploads. ## Resource-Aware Rate Control - Mussel replaced raw request counting with request units (RU), which represent estimated backend work. - RU calculations incorporate: - Fixed per-request overhead. - Rows and payload bytes processed. - Request latency, which distinguishes cached operations from disk-heavy ones. - The system uses calibrated linear formulas for reads and writes, with weights based on compute, network, and disk-I/O measurements. - Dispatchers debit a local token bucket according to each request’s RU cost rather than charging every request equally. - Periodic RU refills preserve simple, static quotas while making them more proportional to actual resource consumption. - Requests are rejected with HTTP 419 when the RU bucket is exhausted. - Load shedding remains separate, allowing latency-based protection to react dynamically without changing the underlying quota-refill mechanism. ## Load Shedding Under Sudden Stress - RU rate limiting smooths normal traffic but may react too slowly to rapidly changing workloads. - Mussel adds a load-shedding layer based on: - Traffic criticality. - A real-time latency ratio. - A CoDel-inspired queue-management policy. - Each dispatcher compares long-term p95 latency with short-term p95 latency. - A ratio near 1.0 indicates stable performance; a drop toward 0.3 signals rapidly increasing latency. - When stress crosses the threshold: - The system raises the effective RU cost for a designated lower-priority client class. - That class’s token bucket drains faster, causing its traffic to back off. - If conditions worsen, the penalty expands to additional classes. - Critical workloads, such as customer support and trust-and-safety traffic, can remain responsive while less important traffic is reduced. - The latency estimate uses the constant-memory P² algorithm, avoiding raw sample storage and cross-node coordination. ## Hot-Key Detection and DDoS Protection - Client-level quotas cannot prevent overload when many clients request the same popular key. - Mussel therefore detects skewed access patterns in real time. - When duplicate requests target a hot key, the system can protect storage by: - Serving responses from cache. - Coalescing identical requests before they reach the backend. - This approach protects the underlying shard whether the traffic comes from legitimate popularity, automation, or a DDoS burst. Mussel’s experience suggests that mature multi-tenant services should move beyond fixed QPS limits. Combining resource-based accounting, priority-aware load shedding, and hot-key mitigation provides a more effective way to preserve reliability while maximizing useful work during unpredictable traffic conditions.

Read original(opens in new tab)
netflixOriginal article

100X Faster: How We Supercharged Netflix Maestro’s Workflow Engine | by Netflix Technology Blog | Netflix TechBlog (opens in new tab)

Netflix has significantly optimized Maestro, its horizontally scalable workflow orchestrator, to meet the evolving demands of low-latency use cases like live events, advertising, and gaming. By redesigning the core engine to transition from a polling-based architecture to a high-performance event-driven model, the team achieved a 100x increase in speed. This evolution reduced workflow overhead from several seconds to mere milliseconds, drastically improving developer productivity and system efficiency. ### Limitations of the Legacy Architecture The original Maestro architecture was built on a three-layer system that, while scalable, introduced significant latency during execution. * **Polling Latency:** The internal flow engine relied on calling execution functions at set intervals, creating a "speedbump" where tasks waited seconds to be picked up by workers. * **Execution Overhead:** The process of translating complex workflow graphs into parallel flows and sequentially chained tasks added internal processing time that hindered sub-hourly and ad-hoc workloads. * **Concurrency Issues:** A lack of strong guarantees from the internal flow engine occasionally led to race conditions, where a single step might be executed by multiple workers simultaneously. ### Transitioning to an Event-Driven Engine To support the highest level of user needs, Netflix replaced the traditional flow engine with a custom, high-performance execution model. * **Direct Dispatching:** The engine moved away from periodic polling in favor of an event-driven mechanism that triggers state transitions instantly. * **State Machine Optimization:** The new design manages the lifecycle of workflows and steps through a more streamlined state machine, ensuring faster transitions between "start," "restart," "stop," and "pause" actions. * **Reduced Data Latency:** The team optimized data access patterns for internal state storage, reducing the time required to write Maestro data to the database during high-volume executions. ### Scalability and Functional Improvements The redesign not only improved speed but also strengthened the engine's ability to handle massive, complex data pipelines. * **Isolation Layers:** The engine maintains strict isolation between the Maestro step runtime (integrated with Spark and Trino) and the underlying execution logic. * **Support for Heterogeneous Workflows:** The supercharged engine continues to support massive workflows with hundreds of thousands of jobs while providing the low latency required for iterative development cycles. * **Reliability Guarantees:** By moving to a more robust internal event bus, the system eliminated the race conditions found in the previous distributed job queue implementation. For organizations managing large-scale Data or ML workflows, moving toward an event-driven orchestration model is essential for supporting sub-hourly execution and low-latency ad-hoc queries. These performance improvements are now available in the Maestro open-source project for wider community adoption.

netflixOriginal article

Building a Resilient Data Platform with Write-Ahead Log at Netflix | by Netflix Technology Blog | Netflix TechBlog (opens in new tab)

Netflix has developed a distributed Write-Ahead Log (WAL) abstraction to address critical data challenges such as accidental corruption, system entropy, and the complexities of cross-region replication. By decoupling data mutation from immediate persistence and providing a unified API, this system ensures strong durability and eventual consistency across diverse storage engines. The WAL acts as a resilient buffer that powers high-leverage features like secondary indexing and delayed retry queues while maintaining the massive scale required for global operations. ### The Role of the WAL Abstraction * The system serves as a centralized mechanism to capture data changes and reliably deliver them to downstream consumers, mitigating the risk of data loss during administrative errors or database corruption. * It provides a simplified `WriteToLog` gRPC endpoint that abstracts underlying infrastructure, allowing developers to focus on data logic rather than the specifics of the storage layer. * By acting as a durable intermediary, it prevents permanent data loss during incidents where primary datastores fail or require schema changes that might otherwise lead to corruption. ### Flexible Personas and Namespaces * The architecture utilizes "namespaces" to define logical separation, allowing different services to configure specific storage backends like Kafka or SQS based on their needs. * The "Delayed Queues" persona leverages SQS to provide a scalable way to retry failed messages in real-time pipelines without sacrificing overall system throughput. * The system can be configured for "Cross-Region Replication," enabling high availability and disaster recovery for storage engines that do not natively support multi-region data transfer. ### Solving System Entropy and Consistency * The WAL addresses the "dual-write" problem, where updates to primary stores (such as Cassandra) and search indices (such as Elasticsearch) can diverge over time, leading to data inconsistency. * It facilitates reliable secondary indexing for NoSQL databases by managing updates to multiple partitions as a coordinated sequence of events. * The platform mitigates operational risks, such as Out-of-Memory (OOM) errors on Key-Value nodes caused by bulk deletes, by staging and throttling mutations through the log. Organizations operating at scale should adopt a WAL-centric architecture to simplify the management of heterogeneous data stores and enhance system resilience. By centralizing the mutation log, teams can implement complex features like Change Data Capture (CDC) and cross-region failover through a single, consistent interface rather than building bespoke solutions for every service.

airbnb4 min readCurated summary

Viaduct, Five Years On: Modernizing the Data-Oriented Service Mesh

Viaduct, Airbnb’s data-oriented service mesh, has evolved substantially over five years while retaining its core model: a central schema, hosted business logic, and re-entrant composition through GraphQL. Its usage has grown eightfold, supporting more than 130 teams and over 1.5 million lines of production code, without increasing operational overhead. Viaduct Modern now aims to simplify its developer API and establish stronger architectural boundaries, alongside the project’s release as open source. ## Adoption and Evolution - Viaduct traffic has increased by a factor of eight since 2020. - More than 130 teams now host code in Viaduct, supported by hundreds of weekly active developers. - The hosted codebase has grown to over 1.5 million lines, with roughly the same amount of test code. - Operational overhead has remained constant, incident-minutes have been cut in half, and costs have grown linearly with QPS. - Viaduct is now available as open-source software. ## Core Principles That Remain - **Central schema:** Viaduct provides one integrated schema connecting domains across Airbnb. - More than 75% of requests are internal. - The schema is developed by many teams but exposed as a connected graph. - **Hosted business logic:** Teams run business logic directly in Viaduct rather than maintaining separate microservices. - This reduces operational overhead and can allow standalone services to be retired. - Viaduct provides a serverless environment so developers can focus on application logic. - **Re-entrancy:** Hosted logic composes with other hosted logic through GraphQL fragments and queries. - This supports modularity. - It helps avoid the tightly coupled structure and maintenance problems associated with traditional monoliths. ## Problems with the Earlier Design - Viaduct’s APIs evolved reactively in response to individual use cases. - Multiple mechanisms emerged for accomplishing similar tasks, creating confusion for developers. - Some capabilities were well supported while others were not. - The framework’s layers had loose, inconsistent interfaces. - The boundary between Viaduct and hosted application code was weak. - These issues made framework improvements increasingly risky because changes could disrupt existing users. ## Simplifying the Tenant API - Viaduct Modern overhauls the developer-facing API and execution engine. - The new Tenant API reduces the implementation choices to two mechanisms: - **Node resolvers** - **Field resolvers** - The choice is determined by the schema rather than by ad hoc behavioral distinctions. - Resolver APIs have been unified wherever possible. - The goal is a smaller, more consistent surface that preserves successful ideas from the old API while removing unnecessary alternatives. ## Tenant Modules and Re-Entrant Composition - Viaduct uses modules and re-entrancy to provide boundaries similar to service definitions and RPC APIs in microservice architectures. - A tenant module combines: - Schema owned by a team - The code implementing that schema - Modules can create rich connections in the shared graph, but direct code dependencies between teams are discouraged. - Instead, teams declare their data requirements through GraphQL fragments and queries. ### Example: Extending the `User` Type - A Core User team owns the base `User` type and resolves fields such as `firstName` and `lastName`. - A Messaging team can extend `User` with a `displayName` field. - Its resolver declares that it needs `firstName` and `lastName`. - Messaging does not depend directly on Core User’s implementation or need to know where those fields originate. - This declarative model lets teams collaborate through the schema while preserving ownership and modularity. ## Framework Modularity - Viaduct Modern also restructures the framework itself. - The system consists of: - The GraphQL execution engine - The Tenant API - Hosted application code - Historically, the interfaces between these layers were weak, making performance and reliability improvements difficult to introduce safely. - The redesign focuses on stronger abstraction boundaries so the framework can evolve independently of application code. Viaduct’s modernization is intended to preserve its centralized, data-oriented model while making development simpler and framework evolution safer. The open-source release provides an opportunity for other organizations to evaluate or adopt this approach to schema-driven, modular service composition.

Read original(opens in new tab)
discord3 min readCurated summary

How Discord Indexes Trillions of Messages

Discord’s original Elasticsearch-based search system worked well for billions of messages but became fragile as message volume and cluster size grew. Redis queues could drop messages, bulk operations failed too broadly, large clusters were difficult to operate, and individual indices could hit Lucene’s roughly two-billion-document limit. Discord’s response was to modernize the platform with Kubernetes, the Elastic Kubernetes Operator, and a multi-cluster “cell” architecture built from smaller clusters. ## The Original Search Architecture - Messages were stored in Elasticsearch indices distributed across two clusters. - Data was sharded by Discord server (guild) or direct message, keeping each guild’s messages together for efficient queries. - Messages were indexed lazily because not every message is searched. - Redis-backed queues supplied workers with message batches for Elasticsearch bulk indexing. ## Problems with the Existing System ### Redis Queue Message Loss - The realtime indexing queue relied on Redis. - When Elasticsearch failures caused the queue to back up, Redis CPU usage could reach its limit. - Once overloaded, Redis began dropping messages, making the indexing pipeline unreliable. ### Fault-Intolerant Bulk Indexing - A batch could contain messages belonging to many different Elasticsearch indices and nodes. - A batch of 50 messages might fan out to dozens of nodes. - If one message failed because its target node was unavailable, Elasticsearch treated the entire bulk request as failed. - All messages were then re-enqueued, increasing queue pressure. - In a 100-node cluster with batches of 50 messages, a single failed node gave each batch roughly a 40% chance of encountering a failure. ### Large-Cluster Overhead - Adding nodes and indices enabled horizontal scaling but increased coordination overhead. - Bulk operations fanned out across more nodes, slowing indexing. - Larger clusters also had a higher probability that some node would fail. ### Difficult Upgrades and Restarts - The system lacked sufficient resilience to individual node outages, making rolling restarts unsafe. - Clusters exceeding 200 nodes and containing terabytes of data would have taken too long to drain gracefully. - Discord therefore remained on outdated operating-system and Elasticsearch versions. - Addressing the Log4Shell vulnerability required taking the entire search system offline while every node was restarted. ### Oversized Indices - Some indices accumulated messages from extremely large guilds. - Each Elasticsearch index is backed by a Lucene index with a limit of approximately two billion documents. - Once that limit was reached, all further indexing failed. - Discord temporarily recovered by identifying and deleting guilds created primarily for message spam, but this was not viable for legitimate high-volume communities. ## Moving Elasticsearch to Kubernetes - Discord chose Kubernetes to improve operational flexibility and resource efficiency. - The Elastic Cloud on Kubernetes (ECK) Operator could define cluster topology and configuration declaratively. - Kubernetes would automate operating-system upgrades. - ECK provided tools for safer rolling restarts and Elasticsearch upgrades. - This marked Discord’s first move toward managing stateful Elasticsearch infrastructure on Kubernetes. ## Smaller Multi-Cluster Cells - Discord planned to replace very large clusters with a larger number of smaller Elasticsearch clusters. - Smaller clusters reduce coordination overhead and limit the impact of individual node failures. - A cell-based design also provides a more manageable scaling and operational boundary than clusters with hundreds of nodes. Discord’s experience demonstrates that scaling Elasticsearch is not only a matter of adding nodes. Reliable operation requires isolating failures, avoiding oversized indices and fan-out-heavy batches, and designing deployment infrastructure that supports upgrades without taking search offline.

Read original(opens in new tab)
datadog3 min readCurated summary

How we use formal modeling, lightweight simulations, and chaos testing to design reliable distributed systems

Courier’s design illustrates why distributed systems need more than unit, integration, and chaos testing. Datadog combined formal modeling, lightweight simulation, and conventional testing to uncover system-level risks before implementation. The approach was especially important after the March 8, 2023 outage, which showed how locally reasonable decisions can produce severe global failures. ## Why Distributed Systems Require Additional Analysis - Distributed systems provide greater scale and availability, but introduce concurrency, coordination, and failure modes that are difficult to reason about intuitively. - Traditional tests operate at relatively low levels of detail and may miss high-level design flaws. - Formal models and simulations allow teams to evaluate system behavior during the design phase, before implementation choices become expensive to change. - Model checking exhaustively explores all states permitted by a design and verifies defined correctness properties. ## Formal Modeling and Lightweight Simulation - Formal modeling uses a high-level specification language to describe: - System components and their interactions - Allowed system states - Properties the system must satisfy - Lightweight simulation builds a replica that runs under controlled conditions to study statistical characteristics such as: - Latency - Cost - Scalability - Behavior under realistic workloads - Modeling verifies correctness but cannot fully assess performance-related concerns. - Neither technique validates the final implementation directly. - Keeping models and simulations synchronized with the production design adds maintenance overhead. - Datadog considered the additional effort worthwhile because Courier was foundational, needed strong reliability guarantees, and incorporated lessons from the 2023 outage. ## Courier’s Requirements Courier was created to replace a decade-old Redis-backed queuing system that had begun to face throughput, scaling, and durability limitations. Its main requirements were: - **Multi-tenancy:** Isolate teams and products so one tenant cannot significantly disrupt others. - **At-least-once delivery:** Messages must not be lost; they must be delivered and acknowledged or sent to a dead-letter queue. - **Graceful degradation and high availability:** Throughput should decline roughly linearly as compute capacity is lost, rather than collapsing entirely. - **Horizontal scalability:** Throughput should increase linearly as compute capacity is added. The graceful-degradation requirement directly addressed the March 8 outage, when lost compute capacity caused a disproportionate throughput failure. ## FoundationDB Sharding for Tenant Isolation - Courier uses multiple FoundationDB clusters. - Each tenant is assigned to a subset of clusters. - No two tenants share the exact same cluster subset. - The initial design used: - Eight FoundationDB clusters - Four clusters per tenant - A theoretical maximum of `8 choose 4 = 70` tenant assignments - If one tenant saturated or disabled its four clusters, other tenants would still retain access to at least 25% of the total cluster capacity. - This arrangement provided sufficient isolation for the intended workloads. ## Broker Layer and High Availability - A broker layer exposes gRPC APIs for: - Sending messages - Receiving messages - Deleting messages - Clients connect only to the brokers, which apply the tenant-sharding logic. - Brokers health-check FoundationDB clusters and remove unhealthy clusters from consideration. - Both brokers and FoundationDB clusters are deployed across three availability zones to improve resilience. Courier demonstrates that formal verification and simulation are valuable complements to implementation testing. For mission-critical distributed services, teams should validate both correctness and operational behavior early, while also using unit, integration, and chaos testing to verify the final system.

Read original(opens in new tab)
datadog3 min readCurated summary

Engineering VP spotlight: Ivo Dimitrov

Ivo Dimitrov’s career evolved from low-level systems programming into engineering leadership focused on large-scale distributed storage. His experience at Microsoft and LinkedIn shaped his approach to building scalable data platforms, while Datadog attracted him with its talented people, modern technology, and culture of experimentation. Today, he leads Datadog’s Distributed Data Systems organization, supporting the company’s metrics, events, query, alerting, and analytics infrastructure. ## From Electrical Engineering to Systems Programming - Dimitrov initially studied electrical engineering and became interested in software while working on digital control systems. - His early work included contributing to a real-time operating system kernel. - He spent roughly a decade as an individual contributor, specializing in: - High-performance systems - Low-level programming - C and C++ - System software ## Transition from Individual Contributor to Manager - At Microsoft, Dimitrov worked on an early version of Azure Blob Storage. - Following a reorganization, he accepted an opportunity to lead his team despite having no prior management experience. - Microsoft supported the transition through: - Leadership mentorship - Formal management training - Guidance on communication, conflict resolution, and interpersonal leadership - He discovered that management allowed him to expand his ownership beyond individual projects and influence broader organizational outcomes. - The role combined his technical background with responsibilities such as cross-functional coordination, team development, and engineering strategy. ## Building Internet-Scale Storage at Microsoft and LinkedIn - At Microsoft, Dimitrov worked on storage systems supporting Hotmail. - After joining LinkedIn in 2014, he adapted to a technology environment centered on open source tools such as MySQL and Java. - He led development of Espresso, LinkedIn’s proprietary key-value storage platform. - The platform matured into a core system supporting approximately 95 percent of LinkedIn’s data sets. - He also helped oversee several other large-scale storage projects: - **Venice**, an open source platform for serving derived data - **Ambry**, an open source blob storage system - **Helix**, an open source cluster manager - These systems supported critical parts of LinkedIn’s internet-scale infrastructure. ## Why Datadog Was Appealing - Dimitrov was drawn to Datadog by three main factors: - Highly capable engineers and leaders - Interesting, modern technology - The opportunity to contribute to a rapidly growing company - Compared with the legacy systems and processes that had accumulated at LinkedIn, Datadog offered less bureaucracy and more freedom to: - Take thoughtful risks - Experiment - Deliver quickly - Fail fast and learn - Iterate and innovate - He was particularly interested in Datadog’s Kubernetes-based Metrics and Events platforms and the challenge of building best-in-class infrastructure during the company’s growth. ## Distributed Data Systems at Datadog - Dimitrov leads the Distributed Data Systems organization, which owns a portfolio of storage and data technologies. - Its responsibilities include: - **Metrics**, supporting metrics and time-series data - **Events**, handling semi-structured data such as logs, profiles, and traces - **Driveline**, a main-memory database optimized for online analytics - The **Cross-Platform Queries** team provides a unified query interface across systems that historically exposed separate, domain-specific APIs. - This reduces the learning curve for engineers and customers. - It abstracts the underlying data stores behind a common API. - The organization also operates Datadog’s Alerts platform, which generates a large share of the queries sent to the Metrics and Events systems. Dimitrov’s experience demonstrates how deep systems expertise can translate into effective engineering leadership. His recommendation by example is to remain technically engaged while expanding one’s scope—from writing individual components to shaping teams, platforms, and long-term engineering direction.

Read original(opens in new tab)
figma2 min readCurated summary

Keeping It 100(x) With Real-time Data At Scale | Figma Blog

Figma’s LiveGraph powers real-time collaboration by subscribing to GraphQL-like queries and updating clients automatically. Rapid growth—tripled sessions since 2021 and fivefold view-request growth in one year—exposed limits in its single-server, mutation-based architecture. Figma launched “LiveGraph 100x,” a redesign focused on scaling reads and database updates while preserving performance and enabling a safe migration. ## LiveGraph’s Role in Figma - LiveGraph keeps data synchronized across collaborative features such as: - File editing - Comments - FigJam voting - It exposes a web API for subscribing to GraphQL-like queries. - Results are returned as JSON trees based on a schema of entities, relationships, and views. - A custom React Hook automatically re-renders interfaces when subscribed data changes. ## The 100x Scaling Initiative Figma’s growing user base increased both the number and cost of LiveGraph client sessions. At the same time, the underlying database evolved from one PostgreSQL instance into vertically and horizontally sharded infrastructure. The redesigned system needed to: - Preserve or improve service-level objectives for initial loads and updates. - Support more database shards reliably and efficiently. - Scale reads and database-update processing independently. - Allow incremental, transparent migrations without disrupting users. ## Limitations of the Original Architecture Originally, LiveGraph consisted of: - A single LiveGraph server. - An in-memory query cache. - One PostgreSQL instance. - A cache that tailed PostgreSQL’s logical replication stream. PostgreSQL writes row mutations to its write-ahead log, including pre- and post-row images and a monotonically increasing sequence number. LiveGraph used these mutations to update cached query results directly rather than recomputing them. This design worked well at smaller scale because: - All updates came from one primary database. - The replication stream provided a global ordering. - Each row mutation could be applied directly to the relevant cached results. ## Sharding Breaks Global Ordering As the original PostgreSQL instance reached capacity, Figma introduced vertical shards and began moving toward broader horizontal scaling. This invalidated the assumption that all database updates arrive in one globally ordered stream. - Multiple shards can generate updates simultaneously. - Their updates have no guaranteed global order. - LiveGraph therefore needed an architecture that could process distributed database changes while maintaining reliable, timely query updates. The growing load made it necessary to rethink LiveGraph fundamentally rather than continue extending its single-database design.

Read original(opens in new tab)
figma3 min readCurated summary

How Figma's Databases Team Lived to Tell the Scale | Figma Blog

Figma’s database stack grew nearly 100× from 2020, pushing its single-Postgres architecture beyond the limits of vertical partitioning. After adding caching, read replicas, and vertically partitioned databases, the team found that individual tables were reaching terabyte and billion-row scales, creating vacuum reliability issues and approaching AWS RDS IOPS limits. The solution was to pursue horizontal sharding while preserving Postgres, minimizing application changes, avoiding massive backfills, and maintaining consistency and rollback options. ## Scaling from One Postgres Database - In 2020, Figma ran on one large Postgres instance. - By the end of 2022, it had introduced: - Caching - Read replicas - Around a dozen vertically partitioned databases - Related tables, such as those for Figma files and organizations, were grouped into separate database partitions. - Vertical partitioning reduced pressure on the system and provided valuable short-term runway. ## Why Vertical Partitioning Was No Longer Enough - The team monitored multiple scaling constraints, including: - CPU and I/O utilization - Table size - Rows written - Database IOPS - Some tables grew to several terabytes and billions of rows. - Large tables began affecting reliability during PostgreSQL vacuum operations, which prevent transaction ID exhaustion. - High-write tables were on track to exceed the maximum IOPS supported by Amazon RDS. - Because a table is the smallest unit of vertical partitioning, splitting databases by table group could not solve these limits. ## Requirements for the Next Scaling Strategy Figma established several design goals for horizontal scaling: - Minimize developer changes and preserve the existing relational data model. - Make future scale-outs transparent to application teams after initial compatibility work. - Avoid months-long backfills of large tables. - Roll out changes incrementally to reduce outage risk. - Preserve rollback capability after physical sharding. - Maintain strong consistency without relying on difficult double-write schemes. - Support near-zero-downtime scale-outs. - Favor technologies and techniques the database team already understood, given the limited runway. ## Evaluating Alternatives - The team considered CockroachDB, TiDB, Spanner, and Vitess. - Moving to another database would have required a risky migration between storage systems while preserving consistency and reliability. - Figma already had substantial operational expertise running Postgres on RDS; replacing it would mean rebuilding that expertise under severe time pressure. - NoSQL systems were also unsuitable because Figma’s application depends on a complex relational data model and requires the flexibility of relational queries. - The team therefore favored a lower-risk approach that retained Postgres and offered greater control over the migration. ## Practical Direction Figma’s experience shows that vertical partitioning can be an effective intermediate step, but it cannot solve limits imposed by individual tables. For systems with rapidly growing relational workloads, horizontal sharding within a familiar database ecosystem can provide a safer path to scale when it is introduced incrementally and designed around consistency, rollback, and minimal application disruption.

Read original(opens in new tab)
datadog3 min readCurated summary

Husky: Exactly-once ingestion and multi-tenancy at scale

Husky, Datadog’s distributed, time-series-oriented event store, is optimized for large scans and aggregations rather than high-volume, low-latency point lookups. This makes exactly-once ingestion challenging, especially at Datadog’s multi-tenant scale. Datadog addresses the problem with deterministic, locality-aware routing that limits deduplication scope, improves storage efficiency, and supports autoscaling ingestion pipelines. ## Husky’s Ingestion Challenge - Husky separates storage and compute, allowing each to scale independently. - Its storage engine is designed primarily for large analytical scans and aggregations. - It is not optimized for massive numbers of low-latency point lookups, complicating duplicate detection during ingestion. - The ingestion system must guarantee that every event is stored exactly once while maintaining: - Multi-tenant scalability - Reasonable ingestion latency - Controlled infrastructure and storage costs ## Routing Events to Storage Shards - Datadog uses an upstream **Shard Router** to introduce locality into Kafka pipelines. - Events are deterministically assigned to shards based on their tenant, timestamp, and event ID. - Each tenant receives a list of shards rather than being permanently assigned to one shard. - A deterministic choice from that list distributes the tenant’s events while keeping the number of active shards as small as practical. - Downstream workers consume one or more shards and perform exactly-once ingestion into Husky. ## Benefits of Data Locality - **Simpler deduplication** - An event with the same timestamp and ID always reaches the same shard. - Deduplication only needs to occur within that shard. - Workers handle smaller sets of event IDs, making in-memory deduplication more efficient. - **Lower storage costs and better performance** - Each shard processes a relatively small set of tenants. - Husky stores each tenant in a separate table and does not mix tenants within files. - More tenants per writer produce more output files, increasing blob-storage costs and compaction work. - Restricting tenant cardinality reduces file creation and improves writer and compactor efficiency. ## Challenges in Deterministic Routing - **Changing shard assignments** - Tenant traffic can increase by one or two orders of magnitude. - Assignments may change when scaling a tenant across more shards or rebalancing traffic among existing shards. - **Distributed router consensus** - Every Shard Router node must make the same routing decision for a given event. - Inconsistent decisions could send duplicates to different shards and undermine exactly-once ingestion. - **Load balancing** - Shards must receive roughly equal traffic so downstream ingestion workers remain balanced. ## Time-Bounded Shard Placements - A simple deterministic mapping can select a shard using a hash of the event ID: ```text shard = shards[hash(event_id) % num_shards] ``` - This approach is cheap and stateless when all routers know the same shard list. - However, changing the shard list can cause the same event ID to map to a different shard, so assignment changes require additional coordination. - The article introduces **time-bounded Shard Placements** to preserve consistent routing while allowing tenant assignments to evolve, though the supplied excerpt ends before explaining the mechanism in detail. Datadog’s core recommendation is to combine deterministic, tenant-aware routing with carefully coordinated assignment changes. This narrows the scope of deduplication while reducing storage overhead and enabling balanced, scalable exactly-once ingestion.

Read original(opens in new tab)