Test `FOR UPDATE` for CRDB

crdb.cs

using Dapper;
using Npgsql;
using System.Diagnostics;

int quantity = 100;

var connStringBuilder = new NpgsqlConnectionStringBuilder();
connStringBuilder.Host = "55.55.55.20";
connStringBuilder.Port = 26257;
connStringBuilder.Username = "root";
connStringBuilder.Database = "nican_test";
connStringBuilder.SslMode = SslMode.Disable;
connStringBuilder.IncludeErrorDetail = true;
connStringBuilder.MaxPoolSize = 150;

await using var conn = new NpgsqlConnection(connStringBuilder.ConnectionString);
await conn.OpenAsync();
await conn.QueryAsync("DROP TABLE IF EXISTS products");
await conn.QueryAsync("CREATE TABLE products (id INT PRIMARY KEY, quantity INT NOT NULL)");
await conn.QueryAsync($"INSERT INTO products(id,quantity) VALUES (1,{quantity})");

var watch = new Stopwatch();
watch.Start();

int workers = 0;

var results = await Task.WhenAll(Enumerable.Range(0, quantity).Select(async id =>
{
    try
    {
        var conn2 = new NpgsqlConnection(connStringBuilder.ConnectionString);
        await conn2.OpenAsync();

        using var trans = await conn2.BeginTransactionAsync();
        await conn2.QueryAsync("SELECT * FROM products WHERE id=1 FOR UPDATE ", transaction: trans);
        int count = Interlocked.Increment(ref workers);

        if(count > 1)
        {
            Console.WriteLine($"Too many workers in the critical zone: Worker#{id}");
        }

        await Task.Delay(TimeSpan.FromMilliseconds(10));
        Interlocked.Decrement(ref workers);

        await conn2.QueryAsync($"UPDATE products SET quantity=quantity-1 WHERE id=1", transaction: trans);
        await trans.CommitAsync();
        return true;
    }
    catch (Exception ex) when (ex.Message.Contains("WriteTooOldError"))
    {
        Console.WriteLine($"Failed id: Worker#{id}");
        return false;
    }
}));

var success = results.Where(t => t).Count();

Console.WriteLine($"Total success: {success}/{quantity}");
Console.WriteLine($"Time taken: {watch.Elapsed.TotalMilliseconds}ms");