using ydb.sdk retry

Author: poma12390

slo/src/Linq2db.Slo/Program.cs

@@ -1,3 +1,4 @@
-using Internal;
+using Linq2db;
+using Internal;
 
-await Cli.Run(new Linq2db.SloTableContext(), args);
+await Cli.Run(new SloTableContext(), args);

slo/src/Linq2db.Slo/SloLinq2DbContext.cs

@@ -1,12 +1,28 @@
-using Internal;
+using LinqToDB;
 using LinqToDB.Data;
+using LinqToDB.Mapping;
+using Internal;
+using LinqToDB.Async;
+using LinqToDB.Internal.DataProvider.Ydb.Internal;
 
 namespace Linq2db;
 
 public sealed class SloTableContext : SloTableContext<SloTableContext.Linq2dbClient>
 {
     protected override string Job => "Linq2DB";
 
+    // ВКЛЮЧАЕМ ретраи SDK глобально для всех DataConnection
+    static SloTableContext()
+    {
+        YdbSdkRetryPolicyRegistration.UseGloballyWithIdempotence(
+            maxAttempts: 10,
+            onRetry: (attempt, ex, delay) =>
+            {
+                // здесь при желании снимать метрики: attempt, delay, ((YdbException)ex).Code
+            }
+        );
+    }
+
     public sealed class Linq2dbClient
     {
         private readonly string _connectionString;
@@ -16,7 +32,8 @@ public Linq2dbClient(string connectionString)
             _connectionString = connectionString;
         }
 
-        public DataConnection Open() => new DataConnection("YDB", _connectionString);
+        public DataConnection Open()
+            => new DataConnection("YDB", _connectionString);
     }
 
     protected override Linq2dbClient CreateClient(Config config)
@@ -27,54 +44,64 @@ protected override async Task Create(Linq2dbClient client, int operationTimeout)
         await using var db = client.Open();
         db.CommandTimeout = operationTimeout;
 
-        // 1) CREATE TABLE
-        await db.ExecuteAsync($@"
-            CREATE TABLE `{SloTable.Name}` (
-                Guid             Uuid,
-                Id               Int32,
-                PayloadStr       Text,
-                PayloadDouble    Double,
-                PayloadTimestamp Timestamp,
-                PRIMARY KEY (Guid, Id)
-            );
-        ");
-
-        await db.ExecuteAsync(SloTable.Options);
+        try
+        {
+            await db.ExecuteAsync($@"
+CREATE TABLE `{SloTable.Name}` (
+    Guid             Uuid,
+    Id               Int32,
+    PayloadStr       Utf8,
+    PayloadDouble    Double,
+    PayloadTimestamp Timestamp,
+    PRIMARY KEY (Guid, Id)
+)");
+        }
+        catch
+        {
+            // YDB не поддерживает IF NOT EXISTS; если таблица есть — окей
+        }
+
+        if (!string.IsNullOrWhiteSpace(SloTable.Options))
+        {
+            await db.ExecuteAsync(SloTable.Options);
+        }
     }
 
-    protected override async Task<int> Save(Linq2dbClient client, SloTable row, int writeTimeout)
+    protected override async Task<int> Save(Linq2dbClient client, SloTable sloTable, int writeTimeout)
     {
         await using var db = client.Open();
         db.CommandTimeout = writeTimeout;
 
         await db.ExecuteAsync($@"
-            UPSERT INTO `{SloTable.Name}` (Guid, Id, PayloadStr, PayloadDouble, PayloadTimestamp)
-            VALUES ({row.Guid}, {row.Id}, {row.PayloadStr}, {row.PayloadDouble}, {row.PayloadTimestamp});
-        ");
+UPSERT INTO `{SloTable.Name}` (Guid, Id, PayloadStr, PayloadDouble, PayloadTimestamp)
+VALUES ({sloTable.Guid}, {sloTable.Id}, {sloTable.PayloadStr}, {sloTable.PayloadDouble}, {sloTable.PayloadTimestamp});
+");
 
-        return 1;
+        return 0;
     }
 
-    protected override async Task<object?> Select(Linq2dbClient client, (Guid Guid, int Id) key, int readTimeout)
+    protected override async Task<object?> Select(Linq2dbClient client, (Guid Guid, int Id) select, int readTimeout)
     {
         await using var db = client.Open();
         db.CommandTimeout = readTimeout;
 
-        var exists = await db.ExecuteAsync<int>($@"
-            SELECT COUNT(*) FROM `{SloTable.Name}` WHERE Guid = {key.Guid} AND Id = {key.Id};
-        ");
-
-        return exists > 0 ? 1 : null;
+        var t = db.GetTable<SloRow>();
+        return await t.FirstOrDefaultAsync(r => r.Guid == select.Guid && r.Id == select.Id);
     }
 
     protected override async Task<int> SelectCount(Linq2dbClient client)
     {
         await using var db = client.Open();
+        return await db.GetTable<SloRow>().CountAsync();
+    }
 
-        var maxId = await db.ExecuteAsync<int?>($@"
-            SELECT MAX(Id) FROM `{SloTable.Name}`;
-        ");
-
-        return maxId ?? 0;
+    [Table(SloTable.Name)]
+    private sealed class SloRow
+    {
+        [Column] public Guid Guid { get; set; }
+        [Column] public int Id { get; set; }
+        [Column] public string?  PayloadStr { get; set; }
+        [Column] public double   PayloadDouble { get; set; }
+        [Column] public DateTime PayloadTimestamp { get; set; }
     }
 }