Introducing Layered Ingress Sharding: Achieving Single-Tenant Isolation in Multi-Tenant Services
Abhishek Tiwari, Vice President of Engineering, Azure Networking Amit Srivastava, Partner Director of PM, Azure Networking Varun Chawla, Partner Director of Engineering, Azure Networking Why Multi‑Tenant Isolation Is Still Hard at Hyperscale Modern cloud platforms thrive on multitenancy. By sharing infrastructure across tenants, services like Azure Front Door (AFD) can deliver massive scale, global reach, and cost efficiency. At hyperscale, however, this efficiency comes with a hard truth: rare failures are inevitable, and their blast radius matters more than their frequency. When hundreds of thousands of tenants share a global data plane, a single misbehaving tenant, configuration regression, or zero-day exploit can turn a low probability event into a high impact outage. Over the years, the industry has developed many protections — rate limiting, circuit breakers, fair share scheduling, crash protection, and various sharding strategies. These techniques dramatically reduce average case risk, but they still struggle to bound worst case impact. In particular, they fall short of delivering what customers intuitively expect: single tenant isolation semantics (the guarantee that one tenant’s failure does not affect another) without requiring dedicated per-tenant infrastructure. At Azure Front Door, we’ve been working on a new architectural approach that directly targets worst case blast radius. Today, I’m excited to introduce Layered Ingress Sharding, a sharding strategy designed to enable single tenant fault isolation for largescale multitenant services. From Traditional Sharding to Ingress Sharding Traditional partitioning assigns each tenant to a fixed shard. This limits blast radius, but tenants in the same shard can still experience complete outages when that shard fails. Shuffle sharding improves on this by assigning each tenant to a subset of instances, where subsets partially overlap, dramatically reducing the probability of widespread impact. However, shuffle sharding still allows 100% availability loss for tenants in the affected shard, relies heavily on client retries, and introduces nontrivial capacity loss in overlapping shards. To address these limitations, we introduced Ingress Sharding. With ingress sharding, an ingress controller, which we call IRIS (Intelligent Routing with Ingress Sharding), sits directly on the data path. Ingress sharding uses shuffle sharding to construct shards from service instances; IRIS operates on top of these shards to perform tenant-aware, capacity-aware routing rather than introducing a new shard construction algorithm. IRIS identifies the tenant for each incoming connection and deterministically maps that tenant to a shard. Instead of relying on clients to retry when a shard is unhealthy, IRIS actively: Monitors the health of service instances Tracks available capacity in real time Retries and reroutes traffic internally Steers traffic away from unhealthy or overloaded instances Dynamically expands the set of service instances used for oversized tenants when sustained load exceeds a single shard’s capacity In effect, IRIS turns shard selection into a real-time, capacity-aware decision that is reevaluated for every new connection. This moves fault resilience and load balancing inside the platform, rather than pushing that burden onto clients. Ingress sharding significantly improves isolation and resilience, but on its own, a tenant can still experience a complete loss of availability if all instances in its shard are affected. Introducing Layers: Isolation Through Independence Layered Sharding adds a new dimension to multitenant isolation. In Layered Sharding, the service is divided into multiple independent layers. A layer represents an independent serving dimension of the system capable of handling tenant traffic independently. This technique was developed to reduce blast radius by changing how tenants are assigned across shards, rather than changing the underlying shard construction within a single layer. A key design goal is that layered sharding is orthogonal to the underlying sharding strategy. It works with existing approaches, whether that is standard partitioning-based sharding, shuffle sharding, or other shard assignment schemes used within a layer. Within each layer: Service instances are grouped into shards using an existing sharding technique (for example, traditional partitioning or shuffle sharding) Each tenant is assigned to a shard independently in that layer Crucially, tenant-to-shard assignments are randomized and independent across layers. This independence is what gives layered sharding its isolation properties, regardless of the specific sharding algorithm used within a single layer. The definition of a layer itself is intentionally flexible and service dependent. In Azure Front Door, each server naturally acts as one layer. In other services, a layer might correspond to a cluster, a scale unit, a fault domain, or even a regional partition — any unit capable of serving tenant traffic independently while preserving uniform load distribution. The result is powerful: even if a tenant’s traffic causes failures in one shard in one layer, it is statistically unlikely that the same tenant will collide with the same peers across many layers. Instead of experiencing a full outage, other tenants see at most a small, transient reduction in capacity, often invisible with standard retry behavior. Layered sharding alone already reduces availability impact across tenants. But when combined with ingress sharding, it enables something fundamentally stronger. Layered Ingress Sharding Layered Ingress Sharding integrates two complementary ideas: Layered Sharding spreads tenants across many independent layers with randomized shard assignments. Ingress Sharding dynamically routes traffic to healthy service instances across layers using real‑time health and capacity signals. The key blast radius reduction that enables single tenant isolation comes from the combination of independent shard randomization across layers with active, intelligent traffic steering. When a tenant misbehaves due to harmful traffic, a bad configuration, or an unknown vulnerability, IRIS detects unhealthy service instances and automatically routes traffic to healthy shards in other layers. Because shard assignments are independent, IRIS can always find unaffected capacity for well-behaved tenants. The resulting behavior is a fundamental shift in multitenant failure dynamics: Outages remain localized to the misbehaving tenant Healthy tenants continue to serve traffic Blast radius shrinks from fleetwide to tenant-local In effect, a shared multi‑tenant system begins to behave like it has single tenant‑ isolation semantics, without abandoning multitenancy. Why the Math Works The guarantees behind layered ingress sharding are not heuristic, they’re statistical. Because tenant-to-shard assignments are randomized independently across layers, the probability that two tenants repeatedly collide in the same shard follows a binomial distribution. With production representative configurations, tens of layers and shuffle-sharded service instances, the probability that an arbitrary tenant experiences a user-visible failure due to another tenant drops below what standard client retries already mask. Instead of asking “Can a noisy neighbor impact me?”, the system answers “What is the probability that a single connection attempt is unlucky across all layers?” and that probability decreases exponentially as the number of layers increases. This allows us to trade catastrophic outages for rare, isolated, and standard retry-mitigated events. A Hidden Benefit: Identifying Bad Tenants Layered ingress sharding provides an additional, powerful side benefit: automated identification of misbehaving tenants. Because shard assignments are computed independently across layers, a tenant that is consistently responsible for failures appears as a common factor across impacted shards and service instances. By correlating signals across layers, the platform can accurately identify the offending tenant and apply targeted mitigations such as isolation, throttling, or traffic steering without relying on coarse-grained circuit breakers that penalize everyone. This dramatically improves response time under high load or adversarial conditions while preserving availability for unaffected tenants. Beyond Azure Front Door While Layered Ingress Sharding was developed in the context of Azure Front Door, the underlying principle is broadly applicable. Any large‑scale multi‑tenant system that: Serves many tenants from shared infrastructure Can distribute traffic uniformly across independent layers Can enforce shard-level isolation within each layer can benefit from this approach. Layers don’t have to be servers they could be clusters, scale units, or regional partitions. The key is independent assignment across layers combined with intelligent ingress routing. We believe this pattern represents a reusable architectural strategy for building resilient, hyperscale, multi‑tenant services. Closing Thoughts Multi‑tenancy doesn’t have to mean shared fate. Layered Ingress Sharding shows that by combining probabilistic isolation with intelligent ingress routing, we can build systems where failures are expected, bounded, and automatically contained, even at hyperscale. Rather than eliminating failure, this approach mathematically constrains its impact. And in large‑scale multi‑tenant platforms, that distinction makes all the difference.Scaling Write Throughput in Azure Database for MySQL Using Application-Level Sharding
This blog post walks through scaling write throughput in Azure Database for MySQL using application level sharding. It starts with the why behind sharding and then builds a complete C# implementation that spreads writes across three Azure Database for MySQL Flexible Servers. Why Shard in the First Place? This post focuses specifically on scaling write throughput. A well-tuned single primary node can take you remarkably far, and techniques such as indexing strategies, write batching, redo log optimization, and vertical compute scaling each deliver real, lasting value. For many workloads, these optimizations are all you will ever need. That said, as write volume continues to grow, a single primary eventually approaches its practical capacity, and at that point the most durable way to keep scaling is to distribute the write workload across multiple primary instances. This architecture is what we call sharding. When you reach this inflection point, there are two primary patterns for managing multiple write nodes: Proxy or Middleware Layer Sharding: A sharding aware proxy sits between the application and a pool of Azure Database for MySQL instances, routing queries based on a shard key. While this abstracts the underlying topology from the application layer, it introduces an additional, complex component to operate, secure, scale, and patch. Application Layer Sharding: The application itself resolves the destination shard key and determines which of the N Azure Database for MySQL instances should receive a write before ever opening a database connection. Each backend target remains a completely standard, independent Azure Database for MySQL instance. This post explores the second approach. The core appeal of application layer sharding is architectural simplicity: it introduces zero infrastructure overhead and eliminates an extra network hop. Every shard behaves exactly like a standalone instance, meaning your existing backup, restore, monitoring pipelines, and the Azure portal function seamlessly without modification. The explicit tradeoff is that you forgo cross shard joins and distributed transactions in exchange for absolute predictability and control over data access patterns. The Plan We will build a small order management service that distributes its data across three Azure Database for MySQL instances that already exist. The application, written in C# on .NET 8, owns the partitioning logic. The premise: the three servers are already provisioned, the firewalls are configured, the network paths are established, and each server has its own administrative credentials. We are not provisioning infrastructure in this post. we are writing the application code that consumes it. mysql-shard-0.mysql.database.azure.com user: shard0_admin pwd: <secret-0> mysql-shard-1.mysql.database.azure.com user: shard1_admin pwd: <secret-1> mysql-shard-2.mysql.database.azure.com user: shard2_admin pwd: <secret-2> Each server hosts an identical appdb database with the same schema: CREATE TABLE users ( user_id BIGINT NOT NULL PRIMARY KEY, email VARCHAR(255) NOT NULL, created_at DATETIME NOT NULL DEFAULT CURRENT_TIMESTAMP, UNIQUE KEY uq_email (email) ); CREATE TABLE orders ( order_id BIGINT NOT NULL PRIMARY KEY, user_id BIGINT NOT NULL, amount_cents INT NOT NULL, created_at DATETIME NOT NULL DEFAULT CURRENT_TIMESTAMP, KEY ix_user (user_id) ); Two design decisions in this schema warrant explanation: No AUTO_INCREMENT for user_id or order_id. Two shards would otherwise generate the same value 42 independently. Instead, we assign identifiers in the application, using a scheme such as Snowflake, ULID, or UUIDv7. orders carries user_id, and we route by it. This is the single most important rule of sharding: choose a shard key that keeps related data colocated, so that the common queries remain on a single shard. A note on UNIQUE KEY uq_email. A unique index enforces uniqueness only within a single physical shard. Because we route by user_id, two users with different IDs and the same email may land on different shards, and both inserts will succeed. If you require globally unique emails, two options exist: (a) maintain a separate email → user_id lookup table on a single "directory" server and write to it first within an idempotent flow, or (b) shard the users table by a hash of email instead. We retain user_id routing throughout this post because it is the correct choice for orders, and we treat per shard email uniqueness as a best effort guard rather than a hard global invariant. How the Partitioning Works The naive approach to sharding is shard = hash(key) % N. This works until you need to add a fourth server, at which point roughly 75% of your data must move. In any system of meaningful size, that is prohibitively expensive. The established solution is virtual buckets. You hash the key into a large, fixed bucket space (here, 1024), then map buckets to physical shards. When you add capacity, you relocate only buckets; you never rehash the entire dataset. In production, the bucket_to_shard_map typically resides in a system such as Azure App Configuration or etcd, so that you can rebalance without redeploying. For this post, we keep it as an in-memory array seeded at startup, which is straightforward to replace later. The Project ShardingDemo/ ├── ShardingDemo.csproj ├── appsettings.json ├── Models.cs ├── ShardRouter.cs ├── UserRepository.cs └── Program.cs ShardingDemo.csproj <Project Sdk="Microsoft.NET.Sdk"> <PropertyGroup> <OutputType>Exe</OutputType> <TargetFramework>net8.0</TargetFramework> <Nullable>enable</Nullable> <ImplicitUsings>enable</ImplicitUsings> </PropertyGroup> <ItemGroup> <PackageReference Include="MySqlConnector" Version="2.6.0" /> <PackageReference Include="Microsoft.Extensions.Hosting" Version="8.0.0" /> <PackageReference Include="Microsoft.Extensions.Configuration.Binder" Version="8.0.0" /> </ItemGroup> <ItemGroup> <Content Include="appsettings.json" CopyToOutputDirectory="PreserveNewest" /> </ItemGroup> </Project> appsettings.json Shards is an ordered list, and a shard's position in the array is its logical ID. { "Shards": [ { "Host": "mysql-shard-0.mysql.database.azure.com", "Database": "appdb", "User": "shard0_admin", "Password": "REPLACE_ME_0" }, { "Host": "mysql-shard-1.mysql.database.azure.com", "Database": "appdb", "User": "shard1_admin", "Password": "REPLACE_ME_1" }, { "Host": "mysql-shard-2.mysql.database.azure.com", "Database": "appdb", "User": "shard2_admin", "Password": "REPLACE_ME_2" } ] } Models.cs namespace ShardingDemo; public sealed record User(long UserId, string Email, DateTime CreatedAt); public sealed record Order(long OrderId, long UserId, int AmountCents, DateTime CreatedAt); public sealed class ShardConfig { public required string Host { get; init; } public required string Database { get; init; } public required string User { get; init; } public required string Password { get; init; } } ShardRouter.cs using System.Security.Cryptography; using System.Text; using MySqlConnector; namespace ShardingDemo; public sealed class Shard : IAsyncDisposable { public int Id { get; } public MySqlDataSource DataSource { get; } public Shard(int id, ShardConfig cfg) { Id = id; var csb = new MySqlConnectionStringBuilder { Server = cfg.Host, Port = 3306, Database = cfg.Database, UserID = cfg.User, Password = cfg.Password, SslMode = MySqlSslMode.Required, Pooling = true, MinimumPoolSize = 2, MaximumPoolSize = 100, ConnectionTimeout = 10, DefaultCommandTimeout = 30, }; DataSource = new MySqlDataSourceBuilder(csb.ConnectionString).Build(); } public ValueTask DisposeAsync() => DataSource.DisposeAsync(); } public sealed class ShardRouter : IAsyncDisposable { private const int VirtualBuckets = 1024; private readonly IReadOnlyList<Shard> _shards; private readonly int[] _bucketToShardId; public ShardRouter(IEnumerable<ShardConfig> configs) { _shards = configs.Select((c, i) => new Shard(i, c)).ToList(); // Even distribution. Replace with a map loaded from your control plane for live rebalancing. _bucketToShardId = new int[VirtualBuckets]; for (int i = 0; i < VirtualBuckets; i++) _bucketToShardId[i] = i % _shards.Count; } public IReadOnlyList<Shard> AllShards => _shards; private static int BucketFor(long shardKey) { byte[] hash = MD5.HashData(Encoding.ASCII.GetBytes(shardKey.ToString())); // Use the first byte pair as an unsigned value, then map it into the bucket space. int value = (hash[0] << 8) | hash[1]; return value % VirtualBuckets; } public Shard ShardForKey(long shardKey) { int bucket = BucketFor(shardKey); return _shards[_bucketToShardId[bucket]]; } public async ValueTask DisposeAsync() { foreach (var s in _shards) await s.DisposeAsync(); } } UserRepository.cs Observe that every per user method calls ShardForKey(userId), even when inserting an order. This is the colocation rule at work. An order and its owning user always reside on the same shard, so queries for a single user only ever reach one shard. Only the cross-shard aggregate (TotalRevenueCentsAsync) must fan out. using MySqlConnector; namespace ShardingDemo; public sealed class UserRepository { private readonly ShardRouter _router; public UserRepository(ShardRouter router) { _router = router; } public async Task CreateUserAsync(long userId, string email, CancellationToken ct = default) { var shard = _router.ShardForKey(userId); await using var conn = await shard.DataSource.OpenConnectionAsync(ct); await using var cmd = conn.CreateCommand(); cmd.CommandText = "INSERT INTO users (user_id, email) VALUES (@id, Email)"; cmd.Parameters.AddWithValue("@id", userId); cmd.Parameters.AddWithValue("@email", email); await cmd.ExecuteNonQueryAsync(ct); } public async Task<User?> GetUserAsync(long userId, CancellationToken ct = default) { var shard = _router.ShardForKey(userId); await using var conn = await shard.DataSource.OpenConnectionAsync(ct); await using var cmd = conn.CreateCommand(); cmd.CommandText = "SELECT user_id, email, created_at FROM users WHERE user_id = ID"; cmd.Parameters.AddWithValue("@id", userId); await using var reader = await cmd.ExecuteReaderAsync(ct); if (!await reader.ReadAsync(ct)) return null; return new User(reader.GetInt64(0), reader.GetString(1), reader.GetDateTime(2)); } public async Task AddOrderAsync(long orderId, long userId, int amountCents, CancellationToken ct = default) { // Routed by user_id, so orders colocate with their owning user. var shard = _router.ShardForKey(userId); await using var conn = await shard.DataSource.OpenConnectionAsync(ct); await using var cmd = conn.CreateCommand(); cmd.CommandText = """ INSERT INTO orders (order_id, user_id, amount_cents) VALUES (@oid, @uid, amt) """; cmd.Parameters.AddWithValue("@oid", orderId); cmd.Parameters.AddWithValue("@uid", userId); cmd.Parameters.AddWithValue("@amt", amountCents); await cmd.ExecuteNonQueryAsync(ct); } public async Task<IReadOnlyList<Order>> GetOrdersForUserAsync(long userId, CancellationToken ct = default) { var shard = _router.ShardForKey(userId); await using var conn = await shard.DataSource.OpenConnectionAsync(ct); await using var cmd = conn.CreateCommand(); cmd.CommandText = """ SELECT order_id, user_id, amount_cents, created_at FROM orders WHERE user_id = @uid """; cmd.Parameters.AddWithValue("@uid", userId); var list = new List<Order>(); await using var reader = await cmd.ExecuteReaderAsync(ct); while (await reader.ReadAsync(ct)) { list.Add(new Order( reader.GetInt64(0), reader.GetInt64(1), reader.GetInt32(2), reader.GetDateTime(3))); } return list; } /// <summary>Cross shard fanout.</summary> public async Task<long> TotalRevenueCentsAsync(CancellationToken ct = default) { var tasks = _router.AllShards.Select(async shard => { await using var conn = await shard.DataSource.OpenConnectionAsync(ct); await using var cmd = conn.CreateCommand(); cmd.CommandText = "SELECT COALESCE(SUM(amount_cents), 0) FROM orders"; var result = await cmd.ExecuteScalarAsync(ct); return Convert.ToInt64(result); }); var perShard = await Task.WhenAll(tasks); return perShard.Sum(); } } Program.cs using Microsoft.Extensions.Configuration; using Microsoft.Extensions.DependencyInjection; using Microsoft.Extensions.Hosting; using ShardingDemo; var builder = Host.CreateApplicationBuilder(args); // Bind Shards:[] from appsettings.json (override with user-secrets / env vars / Key Vault) var shardConfigs = builder.Configuration .GetSection("Shards") .Get<List<ShardConfig>>() ?? throw new InvalidOperationException("No 'Shards' section configured."); if (shardConfigs.Count == 0) throw new InvalidOperationException("At least one shard must be configured."); builder.Services.AddSingleton(_ => new ShardRouter(shardConfigs)); builder.Services.AddSingleton<UserRepository>(); using var host = builder.Build(); var repo = host.Services.GetRequiredService<UserRepository>(); var router = host.Services.GetRequiredService<ShardRouter>(); (long Id, string Email)[] users = { (1001, "ada@example.com"), (2002, "linus@example.com"), (3003, "grace@example.com"), (4004, "alan@example.com"), }; foreach (var (id, email) in users) { await repo.CreateUserAsync(id, email); Console.WriteLine($"user {id} -> shard {router.ShardForKey(id).Id}"); } await repo.AddOrderAsync(orderId: 9001, userId: 1001, amountCents: 4999); await repo.AddOrderAsync(orderId: 9002, userId: 1001, amountCents: 1299); await repo.AddOrderAsync(orderId: 9003, userId: 2002, amountCents: 8800); Console.WriteLine($"\nAda: {await repo.GetUserAsync(1001)}"); Console.WriteLine($"Ada's orders: {(await repo.GetOrdersForUserAsync(1001)).Count}"); Console.WriteLine($"\nTotal revenue across 3 shards: " + $"${await repo.TotalRevenueCentsAsync() / 100m:F2}"); await router.DisposeAsync(); Tracing One Request End to End Consider GetOrdersForUserAsync(1001): ShardForKey(1001) → MD5("1001") → first two bytes as a number → % 1024 → a bucket in the range 0..1023. bucket % 3 → a physical shard → for example mysql-shard-2.mysql.database.azure.com. The MySqlDataSource provides a pooled, TLS encrypted connection authenticated as shard2_admin. The query runs against shard 2's local ix_user index, with no fan out and at single server speed. Every call with userId = 1001, whether GetUser, AddOrder, or GetOrdersForUser, lands on the same shard. That is why orders JOIN users ON orders.user_id = users.user_id WHERE user_id = 1001 executes within a single shard, with no cross-shard traffic. Conclusion The essential point is this. Once a single primary can no longer absorb your write load, sharding becomes a durable answer, and implementing it at the application layer keeps every part of the system explicit and comprehensible. When write volume or dataset size outgrows a single primary, application layer sharding provides several benefits. N independent Azure Database for MySQL instances, each absorbing 1/N of the write traffic. Queries by user that remain on a single shard and behave like an ordinary, modestly sized database. A bucket map approach that allows you to add a fourth, fifth, or Nth shard later by relocating slices of data rather than rehashing the entire dataset. A failure of one shard that affects 1/N of your users rather than all of them. These benefits come at a genuine cost. You must generate identifiers in the application, global uniqueness requires a secondary lookup table, and aggregate queries fan out across shards. A cross shard write, one that must atomically update data on two different shards, can no longer rely on a single database transaction. Instead it needs an orchestrated sequence of local transactions, where each step carries a compensating action that undoes its effect if a later step fails. None of these are insurmountable. They are simply responsibilities you now assume. Sharding is a deliberate step to take only once a single primary has genuinely exhausted its write headroom. When you reach that point, the implementation in this post is a representative blueprint. Stay Connected We welcome your feedback and invite you to share your experiences or suggestions at AskAzureDBforMySQL@service.microsoft.com Thank you for choosing Azure Database for MySQL!
