Cross-Shard Transactions¶

// Inside ShardTransactionParticipant.PrepareAsync()
public async Task<ParticipantVote> PrepareAsync(CancellationToken cancellationToken)
{
    // Execute any pending operations first
    await ExecutePendingOperationsAsync(cancellationToken);

    // Check if we have any changes
    if (!_context.ChangeTracker.HasChanges())
    {
        _vote = ParticipantVote.ReadOnly;
        return _vote;
    }

    // Save changes but don't commit the transaction yet
    // This validates the changes and acquires locks
    await _context.SaveChangesAsync(cancellationToken);

    _vote = ParticipantVote.Prepared;
    return _vote;
}

public class AccountService
{
    private readonly ICrossShardTransactionCoordinator _coordinator;
    private readonly AppDbContext _context;

    public AccountService(
        ICrossShardTransactionCoordinator coordinator,
        AppDbContext context)
    {
        _coordinator = coordinator;
        _context = context;
    }

    public async Task TransferFundsAsync(
        int fromAccountId,
        int toAccountId,
        decimal amount)
    {
        await _coordinator.ExecuteInTransactionAsync(async transaction =>
        {
            // Get source account (may be in shard EU)
            var fromAccount = await _context.Accounts
                .FirstAsync(a => a.Id == fromAccountId);

            // Get destination account (may be in shard US)
            var toAccount = await _context.Accounts
                .FirstAsync(a => a.Id == toAccountId);

            // Perform the transfer
            fromAccount.Balance -= amount;
            toAccount.Balance += amount;

            // SaveChanges coordinates with cross-shard transaction
            await _context.SaveChangesAsync();
        });
    }
}

await using var transaction = await _coordinator.BeginTransactionAsync();

try
{
    // Explicitly enlist shards
    await transaction.EnlistAsync("shard-eu");
    await transaction.EnlistAsync("shard-us");

    // Perform operations
    var euAccount = await GetAccountFromShard("shard-eu", accountId1);
    var usAccount = await GetAccountFromShard("shard-us", accountId2);

    euAccount.Balance -= 1000;
    usAccount.Balance += 1000;

    await SaveToShard("shard-eu", euAccount);
    await SaveToShard("shard-us", usAccount);

    // Commit all shards atomically
    await transaction.CommitAsync();
}
catch
{
    // Rollback on any failure
    await transaction.RollbackAsync();
    throw;
}

builder.Services.AddDbContext<AppDbContext>(options =>
{
    options.UseSqlServer(connectionString);
    options.UseDtde(dtde =>
    {
        // Shards are automatically enrolled in cross-shard transactions
        dtde.AddShard(s => s
            .WithId("shard-eu")
            .WithConnectionString(euConnection));

        dtde.AddShard(s => s
            .WithId("shard-us")
            .WithConnectionString(usConnection));
    });
});

// ICrossShardTransactionCoordinator is automatically registered

var options = new CrossShardTransactionOptions
{
    // Timeout for the entire transaction
    Timeout = TimeSpan.FromMinutes(2),

    // Isolation level
    IsolationLevel = CrossShardIsolationLevel.ReadCommitted,

    // Retry configuration
    EnableRetry = true,
    MaxRetryAttempts = 3,
    RetryDelay = TimeSpan.FromMilliseconds(100),
    UseExponentialBackoff = true,
    MaxRetryDelay = TimeSpan.FromSeconds(10),

    // Optional transaction name for logging/diagnostics
    TransactionName = "FundsTransfer",

    // Enable recovery for long-running transactions
    EnableRecovery = false
};

await _coordinator.ExecuteInTransactionAsync(async tx =>
{
    // Operations here
}, options);

// Default settings - balanced for most use cases
var options = CrossShardTransactionOptions.Default;

// Short-lived transactions - quick timeout, fewer retries
var shortLived = CrossShardTransactionOptions.ShortLived;
// Timeout: 10 seconds, MaxRetryAttempts: 2

// Long-running transactions - extended timeout, recovery enabled
var longRunning = CrossShardTransactionOptions.LongRunning;
// Timeout: 5 minutes, MaxRetryAttempts: 5, EnableRecovery: true

await using var transaction = await _coordinator.BeginTransactionAsync();
// State: Active

await transaction.EnlistAsync("shard-eu");
await transaction.EnlistAsync("shard-us");
// State: Still Active (enlisting doesn't change state)

// Perform operations...
participant.Context.Update(entity);

await transaction.CommitAsync();
// State progression:
//   Active → Preparing → Prepared → Committing → Committed
//                ↓           ↓           ↓
//           (if any abort)  (if failure) (if partial failure)
//                ↓           ↓           ↓
//           RollingBack → RolledBack   Failed

Console.WriteLine(transaction.State); // Committed (or RolledBack/Failed)

public async Task ProcessWithStateHandling()
{
    await using var transaction = await _coordinator.BeginTransactionAsync();

    try
    {
        await transaction.EnlistAsync("shard-eu");
        await transaction.EnlistAsync("shard-us");

        // ... operations ...

        await transaction.CommitAsync();

        // Check final state
        switch (transaction.State)
        {
            case TransactionState.Committed:
                _logger.LogInformation("Transaction {Id} committed successfully",
                    transaction.TransactionId);
                break;

            case TransactionState.Failed:
                // This shouldn't happen if CommitAsync didn't throw,
                // but handle defensively
                _logger.LogError("Transaction {Id} in failed state",
                    transaction.TransactionId);
                break;
        }
    }
    catch (TransactionPrepareException ex)
    {
        // State: RollingBack → RolledBack
        _logger.LogWarning("Prepare failed on shard {Shard}: {Message}",
            ex.FailedShardId, ex.Message);
    }
    catch (TransactionCommitException ex)
    {
        // State: Failed (some committed, some didn't)
        _logger.LogError(
            "CRITICAL: Partial commit! Committed: [{Committed}], Failed: [{Failed}]",
            string.Join(", ", ex.CommittedShards ?? []),
            string.Join(", ", ex.FailedShards ?? []));
    }
    catch (TransactionTimeoutException ex)
    {
        // State: Failed
        _logger.LogError("Transaction timed out after {Timeout}", ex.Timeout);
    }
}

// This is what happens internally
private ShardAnalysisResult AnalyzeChangesForSharding(DbContext context)
{
    // 1. Get all entities with pending changes
    var entries = context.ChangeTracker.Entries()
        .Where(e => e.State is Added or Modified or Deleted);

    // 2. Group by target shard
    var shardGroups = new Dictionary<string, List<EntityEntry>>();

    foreach (var entry in entries)
    {
        var metadata = _metadataRegistry.GetEntityMetadata(entry.Entity.GetType());

        if (metadata?.ShardingConfiguration is null)
        {
            // Non-sharded entity → default shard
            AddToGroup(shardGroups, "_default_", entry);
        }
        else
        {
            // Sharded entity → calculate target shard
            var targetShard = _writeRouter.DetermineTargetShard(entry.Entity);
            AddToGroup(shardGroups, targetShard.ShardId, entry);
        }
    }

    // 3. Determine if cross-shard transaction needed
    return new ShardAnalysisResult
    {
        RequiresCrossShardTransaction = shardGroups.Keys.Count > 1,
        ShardGroups = shardGroups
    };
}

// DTDE automatically detects when entities span multiple shards
_context.Accounts.Add(new Account { Region = "EU", Balance = 5000 });
_context.Accounts.Add(new Account { Region = "US", Balance = 3000 });

// If these entities go to different shards, DTDE automatically
// coordinates a cross-shard transaction
await _context.SaveChangesAsync();

await using var transaction = await _context.Database.BeginTransactionAsync();

// Behind the scenes:
// 1. TransactionStartingAsync() intercepted
// 2. TransparentShardSession created with new cross-shard transaction
// 3. Session tracks all shards touched during this transaction

_context.Accounts.Add(new Account { Region = "EU" });
await _context.SaveChangesAsync();
// → Routed through session to shard-eu participant

_context.Accounts.Add(new Account { Region = "US" });
await _context.SaveChangesAsync();
// → Routed through session to shard-us participant

await transaction.CommitAsync();
// Behind the scenes:
// 1. TransactionCommittingAsync() intercepted
// 2. Session.CommitAsync() called
// 3. Cross-shard 2PC executed for both shards
// 4. Original transaction commit proceeds

public class CrossShardTransactionException : DtdeException
{
    public string? TransactionId { get; }           // e.g., "XS-20241215143052-7f3a2b1c"
    public IReadOnlyCollection<string>? InvolvedShards { get; }  // All shards in transaction
    public IReadOnlyCollection<string>? FailedShards { get; }    // Shards that failed
}

public class TransactionPrepareException : CrossShardTransactionException
{
    public string? FailedShardId { get; }  // The shard that voted Abort
}

// Example: Constraint violation during prepare
try
{
    await _coordinator.ExecuteInTransactionAsync(async tx =>
    {
        await tx.EnlistAsync("shard-eu");

        // This entity violates a unique constraint in shard-eu
        participant.Context.Add(duplicateEntity);

        // PrepareAsync will fail when SaveChanges detects the violation
    });
}
catch (TransactionPrepareException ex)
{
    // Safe! No data was committed anywhere
    _logger.LogWarning(
        "Prepare failed on shard '{Shard}' for transaction {TxId}. " +
        "No data was modified. Safe to retry.",
        ex.FailedShardId,
        ex.TransactionId);
}

public class TransactionCommitException : CrossShardTransactionException
{
    public IReadOnlyCollection<string>? CommittedShards { get; }  // Successfully committed
    // FailedShards inherited - failed to commit
}

// Example: Network failure during commit phase
try
{
    await _coordinator.ExecuteInTransactionAsync(async tx =>
    {
        await tx.EnlistAsync("shard-eu");
        await tx.EnlistAsync("shard-us");

        // ... operations ...

        // Both prepare successfully, but during commit:
        // - shard-eu commits successfully
        // - shard-us network failure!
    });
}
catch (TransactionCommitException ex)
{
    // CRITICAL! Data inconsistency possible
    _logger.LogCritical(
        "PARTIAL COMMIT DETECTED for transaction {TxId}!\n" +
        "  Committed shards: [{Committed}]\n" +
        "  Failed shards: [{Failed}]\n" +
        "  Manual intervention may be required.",
        ex.TransactionId,
        string.Join(", ", ex.CommittedShards ?? []),
        string.Join(", ", ex.FailedShards ?? []));

    // Trigger alerts, create incident ticket, etc.
    await _alertService.RaisePartialCommitAlert(ex);
}

public class TransactionTimeoutException : CrossShardTransactionException
{
    public TimeSpan Timeout { get; }  // The configured timeout duration
}

// Example: Long-running operation
var options = new CrossShardTransactionOptions
{
    Timeout = TimeSpan.FromSeconds(10)
};

try
{
    await _coordinator.ExecuteInTransactionAsync(async tx =>
    {
        await tx.EnlistAsync("shard-eu");

        // This takes too long...
        await Task.Delay(TimeSpan.FromSeconds(15));

    }, options);
}
catch (TransactionTimeoutException ex)
{
    _logger.LogWarning(
        "Transaction {TxId} timed out after {Timeout}. " +
        "Transaction was rolled back.",
        ex.TransactionId,
        ex.Timeout);
}

try
{
    await _coordinator.ExecuteInTransactionAsync(async tx =>
    {
        await tx.EnlistAsync("shard-eu");
        // Modify EU data...

        await tx.EnlistAsync("shard-us");
        // This throws!
        throw new InvalidOperationException("Something went wrong");
    });
}
catch (InvalidOperationException ex)
{
    // Transaction was automatically rolled back
    // EU changes were NOT persisted
    _logger.LogError(ex, "Transfer failed, all changes rolled back");
}

// Internal retry logic
private async Task ExecuteWithRetryAsync(Func<Task> action, CrossShardTransactionOptions options)
{
    var attempts = 0;
    var delay = options.RetryDelay;

    while (true)
    {
        attempts++;

        try
        {
            await action();
            return;
        }
        catch (Exception ex) when (IsTransientError(ex) && attempts < options.MaxRetryAttempts)
        {
            _logger.LogWarning(
                "Transient error on attempt {Attempt}/{Max}. Retrying in {Delay}ms...",
                attempts, options.MaxRetryAttempts, delay.TotalMilliseconds);

            await Task.Delay(delay);

            if (options.UseExponentialBackoff)
            {
                delay = TimeSpan.FromMilliseconds(
                    Math.Min(delay.TotalMilliseconds * 2, options.MaxRetryDelay.TotalMilliseconds));
            }
        }
    }
}

private static bool IsTransientError(Exception ex)
{
    return ex is TimeoutException
        || ex is OperationCanceledException
        || ex.Message.Contains("deadlock", StringComparison.OrdinalIgnoreCase)
        || ex.Message.Contains("timeout", StringComparison.OrdinalIgnoreCase)
        || ex.Message.Contains("connection", StringComparison.OrdinalIgnoreCase);
}

try
{
    await _coordinator.ExecuteInTransactionAsync(async tx =>
    {
        await tx.EnlistAsync("shard-eu");
        await tx.EnlistAsync("shard-us");

        // Operations...

        // During CommitAsync, if EU commits but US fails:
        // 1. EU is already committed (cannot rollback!)
        // 2. US failed to commit
        // 3. TransactionCommitException thrown with details
    });
}
catch (TransactionCommitException ex)
{
    _logger.LogError(ex,
        "Cross-shard transaction failed. TransactionId: {TxId}, " +
        "Committed: [{Committed}], Failed: [{Failed}]",
        ex.TransactionId,
        string.Join(", ", ex.CommittedShards ?? []),
        string.Join(", ", ex.FailedShards ?? []));

    // This is an IN-DOUBT transaction
    // You may need compensating transactions or manual intervention
}

catch (TransactionCommitException ex)
{
    if (ex.CommittedShards?.Contains("shard-eu") == true &&
        ex.FailedShards?.Contains("shard-us") == true)
    {
        // EU committed but US didn't - create compensating transaction
        await _coordinator.ExecuteInTransactionAsync(async tx =>
        {
            await tx.EnlistAsync("shard-eu");

            // Reverse the EU changes
            var euAccount = await GetAccount("shard-eu", fromAccountId);
            euAccount.Balance += transferAmount; // Reverse the debit

            // Log the compensation
            participant.Context.Add(new CompensationLog
            {
                OriginalTransactionId = ex.TransactionId,
                Reason = "US shard commit failed",
                CompensatedAt = DateTime.UtcNow
            });
        });
    }
}

// Instead of cross-shard transaction, use outbox
public async Task TransferWithOutbox(int fromId, int toId, decimal amount)
{
    // Step 1: Debit source account and write to outbox (single shard)
    await _coordinator.ExecuteInTransactionAsync(async tx =>
    {
        await tx.EnlistAsync("shard-eu");

        var fromAccount = await GetAccount(fromId);
        fromAccount.Balance -= amount;

        // Outbox message - will be processed later
        participant.Context.Add(new OutboxMessage
        {
            Type = "CreditAccount",
            Payload = JsonSerializer.Serialize(new { toId, amount }),
            CreatedAt = DateTime.UtcNow
        });
    });

    // Step 2: Background processor reads outbox and credits destination
    // This is eventually consistent but avoids 2PC risks
}

public class TransferSaga
{
    public async Task Execute(TransferRequest request)
    {
        var sagaId = Guid.NewGuid().ToString();

        try
        {
            // Step 1: Reserve funds in source
            await ReserveFunds(request.FromAccountId, request.Amount, sagaId);

            // Step 2: Credit destination
            await CreditAccount(request.ToAccountId, request.Amount, sagaId);

            // Step 3: Confirm reservation (debit source)
            await ConfirmReservation(request.FromAccountId, sagaId);
        }
        catch
        {
            // Compensate all completed steps
            await CompensateSaga(sagaId);
            throw;
        }
    }
}

// ✅ Good - minimize time in transaction
await _coordinator.ExecuteInTransactionAsync(async tx =>
{
    // Quick read and update
    var account = await _context.Accounts.FindAsync(id);
    account.Balance += amount;
    await _context.SaveChangesAsync();
});

// ❌ Bad - long-running operations in transaction
await _coordinator.ExecuteInTransactionAsync(async tx =>
{
    await SomeSlowExternalApiCall(); // Don't do this!
    var account = await _context.Accounts.FindAsync(id);
    account.Balance += amount;
    await _context.SaveChangesAsync();
});

// ✅ Good - minimize shards involved
await _coordinator.ExecuteInTransactionAsync(async tx =>
{
    await tx.EnlistAsync("shard-eu");
    // Operations on EU only
});

// ⚠️ Caution - many shards increases failure probability
await _coordinator.ExecuteInTransactionAsync(async tx =>
{
    await tx.EnlistAsync("shard-1");
    await tx.EnlistAsync("shard-2");
    await tx.EnlistAsync("shard-3");
    await tx.EnlistAsync("shard-4");
    // More shards = higher chance of partial failure
});

// ✅ Good - idempotent (can safely retry)
await _coordinator.ExecuteInTransactionAsync(async tx =>
{
    var transfer = await _context.Transfers
        .FirstOrDefaultAsync(t => t.IdempotencyKey == key);

    if (transfer != null) return; // Already processed

    // Process transfer...
});

// ❌ Bad - not idempotent
await _coordinator.ExecuteInTransactionAsync(async tx =>
{
    account.Balance += amount; // Retries would add multiple times!
});

var options = new CrossShardTransactionOptions
{
    Timeout = TimeSpan.FromSeconds(30)
};

try
{
    await _coordinator.ExecuteInTransactionAsync(async tx =>
    {
        // Operations...
    }, options);
}
catch (TimeoutException)
{
    _logger.LogWarning("Transaction timed out - check for deadlocks");
    // Consider: longer timeout, smaller transaction, or async processing
}

// Transaction A                    // Transaction B
account.Balance = 1000;
await context.SaveChangesAsync();   // Balance = 1000 (uncommitted)
                                    var bal = account.Balance; // Reads 1000!
await transaction.RollbackAsync();  // Oops, rolled back!
                                    // Transaction B now has WRONG data

// Transaction A                    // Transaction B
var bal1 = account.Balance; // 500
                                    account.Balance = 1000;
                                    await context.SaveChangesAsync();
                                    await transaction.CommitAsync();
var bal2 = account.Balance; // 1000 (different!)

// Transaction A                    // Transaction B
var count1 = await context.Accounts
    .Where(a => a.Region == "EU")
    .CountAsync(); // Returns 10
                                    context.Add(new Account { Region = "EU" });
                                    await context.SaveChangesAsync();
                                    await transaction.CommitAsync();
var count2 = await context.Accounts
    .Where(a => a.Region == "EU")
    .CountAsync(); // Returns 11 (phantom!)

// ReadCommitted (Default) - General purpose
var defaultOptions = new CrossShardTransactionOptions
{
    IsolationLevel = CrossShardIsolationLevel.ReadCommitted
};

// Serializable - Critical financial operations
var financialOptions = new CrossShardTransactionOptions
{
    IsolationLevel = CrossShardIsolationLevel.Serializable,
    Timeout = TimeSpan.FromSeconds(30) // Shorter timeout due to more locking
};

await _coordinator.ExecuteInTransactionAsync(async tx =>
{
    // All participants will use Serializable isolation
    await tx.EnlistAsync("shard-eu");
    await tx.EnlistAsync("shard-us");

    // Range locks prevent phantom reads
    // Row locks prevent non-repeatable reads
    // Uncommitted changes invisible
}, financialOptions);

var snapshotOptions = new CrossShardTransactionOptions
{
    IsolationLevel = CrossShardIsolationLevel.Snapshot
};

await _coordinator.ExecuteInTransactionAsync(async tx =>
{
    // Long-running report - won't block other operations
    var euAccounts = await GetAllAccounts("shard-eu");
    var usAccounts = await GetAllAccounts("shard-us");

    // Even if accounts are updated during this report,
    // we see a consistent snapshot from when we started

    var report = GenerateReport(euAccounts, usAccounts);

}, snapshotOptions);

// Internal mapping to System.Data.IsolationLevel
private static IsolationLevel MapIsolationLevel(CrossShardIsolationLevel level) => level switch
{
    CrossShardIsolationLevel.ReadCommitted => IsolationLevel.ReadCommitted,
    CrossShardIsolationLevel.RepeatableRead => IsolationLevel.RepeatableRead,
    CrossShardIsolationLevel.Serializable => IsolationLevel.Serializable,
    CrossShardIsolationLevel.Snapshot => IsolationLevel.Snapshot,
    _ => IsolationLevel.ReadCommitted
};

// Named transactions for easier identification
var options = new CrossShardTransactionOptions
{
    TransactionName = "FundsTransfer"
};

await _coordinator.ExecuteInTransactionAsync(async tx =>
{
    Console.WriteLine(tx.TransactionId);
    // Output: XS-FundsTransfer-20241215143052-7f3a2b1c
}, options);

builder.Services.AddLogging(logging =>
{
    // Transaction coordinator logs
    logging.AddFilter("Dtde.Core.Transactions", LogLevel.Debug);

    // EF Core interceptor logs
    logging.AddFilter("Dtde.EntityFramework.Infrastructure", LogLevel.Debug);

    // For production, use Information level
    // logging.AddFilter("Dtde.Core.Transactions", LogLevel.Information);
});

public class MonitoredAccountService
{
    private readonly ICrossShardTransactionCoordinator _coordinator;
    private readonly ILogger<MonitoredAccountService> _logger;

    public async Task TransferWithMonitoring(TransferRequest request)
    {
        var options = new CrossShardTransactionOptions
        {
            TransactionName = $"Transfer-{request.FromAccount}-to-{request.ToAccount}"
        };

        using var scope = _logger.BeginScope(new Dictionary<string, object>
        {
            ["FromAccount"] = request.FromAccount,
            ["ToAccount"] = request.ToAccount,
            ["Amount"] = request.Amount,
            ["CorrelationId"] = request.CorrelationId
        });

        var stopwatch = Stopwatch.StartNew();

        try
        {
            await _coordinator.ExecuteInTransactionAsync(async tx =>
            {
                _logger.LogDebug(
                    "Starting transfer transaction {TxId}",
                    tx.TransactionId);

                // ... operations ...

                _logger.LogDebug(
                    "Enlisted shards: {Shards}",
                    string.Join(", ", tx.EnlistedShards));

            }, options);

            stopwatch.Stop();

            _logger.LogInformation(
                "Transfer completed successfully in {Duration}ms",
                stopwatch.ElapsedMilliseconds);
        }
        catch (TransactionPrepareException ex)
        {
            _logger.LogWarning(ex,
                "Transfer prepare failed on shard {Shard}",
                ex.FailedShardId);
            throw;
        }
        catch (TransactionCommitException ex)
        {
            _logger.LogCritical(ex,
                "PARTIAL COMMIT: Committed={Committed}, Failed={Failed}",
                string.Join(",", ex.CommittedShards ?? []),
                string.Join(",", ex.FailedShards ?? []));
            throw;
        }
    }
}

public class TransactionMetrics
{
    // Counters
    public int TransactionsStarted { get; set; }
    public int TransactionsCommitted { get; set; }
    public int TransactionsRolledBack { get; set; }
    public int TransactionsFailed { get; set; }
    public int TransactionsTimedOut { get; set; }
    public int RetriesPerformed { get; set; }

    // Histograms
    public TimeSpan AverageDuration { get; set; }
    public TimeSpan P95Duration { get; set; }
    public TimeSpan P99Duration { get; set; }

    // Gauges
    public int ActiveTransactions { get; set; }
    public int AverageShardsPerTransaction { get; set; }
}

// Example: Adding OpenTelemetry tracing
builder.Services.AddOpenTelemetry()
    .WithTracing(tracing =>
    {
        tracing.AddSource("Dtde.Core.Transactions");
        tracing.AddSource("Dtde.EntityFramework.Infrastructure");
    });

public class CrossShardTransactionHealthCheck : IHealthCheck
{
    private readonly ICrossShardTransactionCoordinator _coordinator;
    private readonly IShardRegistry _shardRegistry;

    public async Task<HealthCheckResult> CheckHealthAsync(
        HealthCheckContext context,
        CancellationToken cancellationToken)
    {
        try
        {
            // Quick test transaction
            var options = CrossShardTransactionOptions.ShortLived;
            options.TransactionName = "HealthCheck";

            await _coordinator.ExecuteInTransactionAsync(async tx =>
            {
                foreach (var shard in _shardRegistry.GetAllShards().Take(2))
                {
                    await tx.EnlistAsync(shard.ShardId, cancellationToken);
                }
                // No actual changes - just testing coordination
            }, options, cancellationToken);

            return HealthCheckResult.Healthy("Cross-shard transactions operational");
        }
        catch (Exception ex)
        {
            return HealthCheckResult.Unhealthy(
                "Cross-shard transaction test failed",
                ex);
        }
    }
}

// ❌ Inefficient - multiple cross-shard transactions
foreach (var transfer in transfers)
{
    await _coordinator.ExecuteInTransactionAsync(async tx =>
    {
        await ProcessTransfer(transfer);
    });
}

// ✅ Efficient - single cross-shard transaction
await _coordinator.ExecuteInTransactionAsync(async tx =>
{
    foreach (var transfer in transfers)
    {
        await ProcessTransfer(transfer);
    }
});

// Design entities to minimize cross-shard operations
public class Account
{
    public int Id { get; set; }
    public string Region { get; set; }  // Shard key

    // Keep related data in same shard
    public List<Transaction> Transactions { get; set; }
    public List<AuditLog> AuditLogs { get; set; }
}

// For read-only operations, skip 2PC overhead
if (isReadOnly)
{
    // Use direct queries to each shard (no transaction coordination)
    var euData = await QueryShard("shard-eu");
    var usData = await QueryShard("shard-us");
    return Merge(euData, usData);
}
else
{
    // Only use cross-shard transaction for writes
    await _coordinator.ExecuteInTransactionAsync(async tx =>
    {
        // ... write operations ...
    });
}

await _coordinator.ExecuteInTransactionAsync(async tx =>
{
    // This will throw InvalidOperationException!
    await _coordinator.ExecuteInTransactionAsync(async innerTx =>
    {
        // Nested transaction not supported
    });
});

await using var transaction = await _coordinator.BeginTransactionAsync();

// Get or create participant for a shard
var crossShardTx = (CrossShardTransaction)transaction;
var participant = await crossShardTx.GetOrCreateParticipantAsync("shard-eu");

// Access the participant's DbContext directly
participant.Context.Add(new Account { Region = "EU", Balance = 1000 });

// Queue operations for later execution
participant.EnqueueOperation(async ctx =>
{
    var account = await ctx.Set<Account>().FirstAsync();
    account.Balance += 500;
});

// Execute all queued operations
await participant.ExecutePendingOperationsAsync();

// Check participant state
Console.WriteLine($"Has changes: {participant.HasPendingChanges}");
Console.WriteLine($"Operation count: {participant.PendingOperationCount}");
Console.WriteLine($"Vote: {participant.Vote}"); // Pending until prepare

await transaction.CommitAsync();
Console.WriteLine($"Vote: {participant.Vote}"); // Prepared (after successful prepare)

public class CustomShardContextFactory : IShardContextFactory
{
    public async Task<DbContext> CreateContextAsync(
        string shardId,
        CancellationToken cancellationToken)
    {
        var connectionString = GetConnectionString(shardId);

        var optionsBuilder = new DbContextOptionsBuilder<AppDbContext>();
        optionsBuilder.UseSqlServer(connectionString);

        // Add custom configuration per shard
        if (shardId.StartsWith("archive-"))
        {
            optionsBuilder.UseQueryTrackingBehavior(QueryTrackingBehavior.NoTracking);
        }

        return new AppDbContext(optionsBuilder.Options);
    }
}

// Check if there's an active transaction
if (_coordinator.HasActiveTransaction)
{
    var currentTx = _coordinator.CurrentTransaction;
    Console.WriteLine($"Active transaction: {currentTx?.TransactionId}");
}

// The ambient transaction flows across async calls
await _coordinator.ExecuteInTransactionAsync(async tx =>
{
    // CurrentTransaction is available throughout this scope
    await SomeNestedMethod();
});

private async Task SomeNestedMethod()
{
    // Can access the ambient transaction
    var tx = _coordinator.CurrentTransaction;
    Console.WriteLine($"In nested method, transaction: {tx?.TransactionId}");
}

// Inside CrossShardTransaction constructor
_timeoutCts = new CancellationTokenSource(Timeout);
_timeoutCts.Token.Register(() => OnTimeout());

// Timeout callback
private void OnTimeout()
{
    if (State is TransactionState.Committed or TransactionState.RolledBack)
        return;

    _logger.LogWarning("Transaction {Id} timed out in state {State}",
        TransactionId, State);

    State = TransactionState.Failed;
}

// During operations, the timeout is linked
using var linkedCts = CancellationTokenSource.CreateLinkedTokenSource(
    cancellationToken, _timeoutCts.Token);

await operation(linkedCts.Token);