EFCore Sql Server tests

.github/workflows/ci.yml

@@ -149,6 +149,41 @@ jobs:
         path: |
           **/TestResults/*
           **/logs/*
+  test-efcore-sqlserver:
+    name: Microsoft Entity Framework Core SQL Server provider tests
+    runs-on: ubuntu-latest
+    strategy:
+      matrix:
+        provider: [ "EFCore-SqlServer" ]
+    services:
+      mssql:
+        image: mcr.microsoft.com/mssql/server:latest
+        ports:
+          - 1433:1433
+        env:
+          ACCEPT_EULA: "Y"
+          MSSQL_PID: "Developer"
+          # [SuppressMessage("Microsoft.Security", "CS002:SecretInNextLine", Justification="False positive")]
+          SA_PASSWORD: "yourStrong(!)Password"
+    steps:
+      - uses: actions/checkout@v2
+      - name: Setup .NET
+        uses: actions/setup-dotnet@v1
+        with:
+          dotnet-version: |
+            3.1.x
+            7.0.x
+      - name: Test
+        run: dotnet test --filter "Category=${{ matrix.provider }}&(Category=BVT|Category=SlowBVT|Category=Functional)" --blame-hang-timeout 10m --logger "trx" -- -parallel none -noshadow
+      - name: Archive Test Results
+        if: always()
+        uses: actions/upload-artifact@v3
+        with:
+          name: test_output
+          retention-days: 1
+          path: |
+            **/TestResults/*
+            **/logs/*
   test-sqlserver:
     name: Microsoft SQL Server provider tests
     runs-on: ubuntu-latest

src/EFCore/Orleans.Clustering.EntityFrameworkCore/EFMembershipTable.cs

@@ -97,7 +97,7 @@ public async Task CleanupDefunctSiloEntries(DateTimeOffset beforeDate)
             var silos = await ctx.Silos
                 .Where(s =>
                     s.ClusterId == this._clusterId &&
-                    s.Status == SiloStatus.Dead &&
+                    s.Status != SiloStatus.Active &&
                     s.IAmAliveTime < beforeDate)
                 .ToArrayAsync()
                 .ConfigureAwait(false);
@@ -122,7 +122,7 @@ public async Task<MembershipTableData> ReadRow(SiloAddress key)
         {
             var ctx = this._dbContextFactory.CreateDbContext();
 
-            var record = await ctx.Silos.Include(s => s.ClusterId).AsNoTracking()
+            var record = await ctx.Silos.Include(s => s.Cluster).AsNoTracking()
                 .SingleOrDefaultAsync(s =>
                     s.ClusterId == this._clusterId &&
                     s.Address == key.Endpoint.Address.ToString() &&
@@ -137,7 +137,7 @@ public async Task<MembershipTableData> ReadRow(SiloAddress key)
 
             var version = new TableVersion(
                 record.Cluster.Version,
-                BitConverter.ToUInt64(record.ETag).ToString()
+                BitConverter.ToUInt64(record.Cluster.ETag).ToString()
             );
 
             var memEntries = new List<Tuple<MembershipEntry, string>> {Tuple.Create(ConvertRecord(record), BitConverter.ToUInt64(record.ETag).ToString())};
@@ -213,8 +213,9 @@ public async Task<bool> InsertRow(MembershipEntry entry, TableVersion tableVersi
             await ctx.SaveChangesAsync().ConfigureAwait(false);
             return true;
         }
-        catch (DbUpdateConcurrencyException)
+        catch (DbUpdateException exc)
         {
+            this._logger.LogWarning(exc, "Failure inserting entry for cluster {Cluster}", this._clusterId);
             return false;
         }
         catch (Exception exc)
@@ -231,7 +232,6 @@ public async Task<bool> UpdateRow(MembershipEntry entry, string etag, TableVersi
         {
             var clusterRecord = this.ConvertToRecord(tableVersion);
             var siloRecord = this.ConvertToRecord(entry);
-            siloRecord.ClusterId = clusterRecord.Id;
             siloRecord.ETag = BitConverter.GetBytes(ulong.Parse(etag));
 
             var ctx = this._dbContextFactory.CreateDbContext();
@@ -359,8 +359,14 @@ private SiloRecord ConvertToRecord(in MembershipEntry memEntry)
         return record;
     }
 
-    private ClusterRecord ConvertToRecord(in TableVersion tableVersion)
+    private ClusterRecord ConvertToRecord( in TableVersion tableVersion)
     {
-        return new() {Id = this._clusterId, Version = tableVersion.Version, Timestamp = DateTimeOffset.UtcNow, ETag = BitConverter.GetBytes(ulong.Parse(tableVersion.VersionEtag))};
+        return new()
+        {
+            Id = this._clusterId,
+            Version = tableVersion.Version,
+            Timestamp = DateTimeOffset.UtcNow,
+            ETag = BitConverter.GetBytes(ulong.Parse(tableVersion.VersionEtag))
+        };
     }
 }

src/EFCore/Orleans.Clustering.EntityFrameworkCore/Orleans.Clustering.EntityFrameworkCore.csproj

@@ -11,6 +11,7 @@
 
   <ItemGroup>
     <InternalsVisibleTo Include="Orleans.Clustering.EntityFrameworkCore.SqlServer"/>
+    <InternalsVisibleTo Include="Tester.EFCore"/>
   </ItemGroup>
 
   <PropertyGroup>

src/EFCore/Orleans.GrainDirectory.EntityFrameworkCore/EFCoreGrainDirectory.cs

@@ -47,14 +47,14 @@ public EFCoreGrainDirectory(
                         c.GrainId == grainIdStr)
                     .ConfigureAwait(false);
 
-                var previousRecord = this.FromGrainAddress(previousAddress);
+                var previousEntry = this.FromGrainAddress(previousAddress);
 
                 if (record is null)
                 {
                     ctx.Activations.Add(toRegister);
                     await ctx.SaveChangesAsync().ConfigureAwait(false);
                 }
-                else if (record.ActivationId != previousRecord.ActivationId || record.SiloAddress != previousRecord.SiloAddress)
+                else if (record.ActivationId != previousEntry.ActivationId || record.SiloAddress != previousEntry.SiloAddress)
                 {
                     return await Lookup(address.GrainId).ConfigureAwait(false);
                 }
@@ -65,7 +65,7 @@ public EFCoreGrainDirectory(
                     ctx.Activations.Update(toRegister);
                     await ctx.SaveChangesAsync().ConfigureAwait(false);
 
-                    return address;
+                    return this.ToGrainAddress(toRegister);
                 }
             }
             else
@@ -74,11 +74,10 @@ public EFCoreGrainDirectory(
                 await ctx.SaveChangesAsync().ConfigureAwait(false);
             }
         }
-        catch (Exception exc)
+        catch
         {
-            this._logger.LogWarning(exc, "Unable to update Grain Directory");
-            WrappedException.CreateAndRethrow(exc);
-            throw;
+            // Possible race condition?
+            return await Lookup(address.GrainId).ConfigureAwait(false);
         }
 
         return await Lookup(address.GrainId).ConfigureAwait(false);

src/EFCore/Orleans.Persistence.EntityFrameworkCore/EFGrainStorage.cs

@@ -15,6 +15,7 @@ namespace Orleans.Persistence.EntityFrameworkCore;
 
 internal class EFGrainStorage<TDbContext> : IGrainStorage, ILifecycleParticipant<ISiloLifecycle> where TDbContext : GrainStateDbContext<TDbContext>
 {
+    private const string ANY_ETAG = "*";
     private readonly ILogger _logger;
     private readonly string _name;
     private readonly string _serviceId;
@@ -25,15 +26,14 @@ public EFGrainStorage(
         string name,
         ILoggerFactory loggerFactory,
         IOptions<ClusterOptions> clusterOptions,
+        IDbContextFactory<TDbContext> dbContextFactory,
         IServiceProvider serviceProvider)
     {
         this._serviceProvider = serviceProvider;
         this._name = name;
         this._serviceId = clusterOptions.Value.ServiceId;
         this._logger = loggerFactory.CreateLogger<EFGrainStorage<TDbContext>>();
-
-        var dbContextFactory = this._serviceProvider.GetService<IDbContextFactory<TDbContext>>();
-        this._dbContextFactory = dbContextFactory ?? throw new OrleansConfigurationException("There are no GrainStateDbContext registered");
+        this._dbContextFactory = dbContextFactory;
     }
 
     public async Task ReadStateAsync<T>(string stateName, GrainId grainId, IGrainState<T> grainState)
@@ -91,14 +91,31 @@ public async Task WriteStateAsync<T>(string stateName, GrainId grainId, IGrainSt
             Data = JsonSerializer.Serialize(grainState.State),
         };
 
-        if (grainState.RecordExists)
+        if (string.IsNullOrWhiteSpace(grainState.ETag))
         {
-            record.ETag = BitConverter.GetBytes(ulong.Parse(grainState.ETag));
-            ctx.GrainState.Update(record);
+            ctx.GrainState.Add(record);
+        }
+        else if (grainState.ETag == ANY_ETAG)
+        {
+            var etag = await ctx.GrainState.AsNoTracking().Where(r =>
+                    r.ServiceId == this._serviceId &&
+                    r.GrainType == grainType &&
+                    r.StateType == stateName &&
+                    r.GrainId == id)
+                .Select(r => r.ETag)
+                .FirstOrDefaultAsync();
+
+            if (etag is not null)
+            {
+                record.ETag = etag;
+            }
+
+            ctx.Update(record);
         }
         else
         {
-            ctx.GrainState.Add(record);
+            record.ETag = BitConverter.GetBytes(ulong.Parse(grainState.ETag));
+            ctx.GrainState.Update(record);
         }
 
         try

src/EFCore/Orleans.Persistence.EntityFrameworkCore/Orleans.Persistence.EntityFrameworkCore.csproj

@@ -11,6 +11,7 @@
 
   <ItemGroup>
     <InternalsVisibleTo Include="Orleans.Persistence.EntityFrameworkCore.SqlServer"/>
+    <InternalsVisibleTo Include="Tester.EFCore"/>
   </ItemGroup>
 
   <PropertyGroup>

src/EFCore/Orleans.Reminders.EntityFrameworkCore/EFReminderTable.cs

@@ -119,7 +119,30 @@ public async Task<string> UpsertRow(ReminderEntry entry)
 
             var ctx = this._dbContextFactory.CreateDbContext();
 
-            ctx.Reminders.Update(record);
+            if (string.IsNullOrWhiteSpace(entry.ETag))
+            {
+                var foundRecord = await ctx.Reminders
+                    .AsNoTracking()
+                    .SingleOrDefaultAsync(r =>
+                        r.ServiceId == this._serviceId &&
+                        r.Name == entry.ReminderName &&
+                        r.GrainId == entry.GrainId.ToString())
+                    .ConfigureAwait(false);
+
+                if (foundRecord is not null)
+                {
+                    record.ETag = foundRecord.ETag;
+                    ctx.Reminders.Update(record);
+                }
+                else
+                {
+                    ctx.Reminders.Add(record);
+                }
+            }
+            else
+            {
+                ctx.Reminders.Update(record);
+            }
 
             await ctx.SaveChangesAsync().ConfigureAwait(false);
 
@@ -197,16 +220,22 @@ public async Task TestOnlyClearTable()
 
     private ReminderRecord ConvertToRecord(ReminderEntry entry)
     {
-        return new ReminderRecord
+        var record = new ReminderRecord
         {
             ServiceId = this._serviceId,
             GrainHash = entry.GrainId.GetUniformHashCode(),
             GrainId = entry.GrainId.ToString(),
             Name = entry.ReminderName,
             Period = entry.Period,
-            StartAt = entry.StartAt,
-            ETag = BitConverter.GetBytes(ulong.Parse(entry.ETag))
+            StartAt = entry.StartAt
         };
+
+        if (!string.IsNullOrWhiteSpace(entry.ETag))
+        {
+            record.ETag = BitConverter.GetBytes(ulong.Parse(entry.ETag));
+        }
+
+        return record;
     }
 
     private ReminderEntry ConvertToEntity(ReminderRecord record)

src/Orleans.Core/Properties/AssemblyInfo.cs

@@ -16,6 +16,7 @@
 [assembly: InternalsVisibleTo("Tester")]
 [assembly: InternalsVisibleTo("Tester.AzureUtils")]
 [assembly: InternalsVisibleTo("Tester.Cosmos")]
+[assembly: InternalsVisibleTo("Tester.EFCore")]
 [assembly: InternalsVisibleTo("Tester.AdoNet")]
 [assembly: InternalsVisibleTo("Tester.Redis")]
 [assembly: InternalsVisibleTo("Tester.ZooKeeperUtils")]

test/Extensions/Tester.EFCore/CollectionFixture.cs

@@ -1,5 +1,4 @@
 using TestExtensions;
-using Xunit;
 
 namespace Tester.EFCore;
 

test/Extensions/Tester.EFCore/EFCoreFixture.cs

@@ -0,0 +1,76 @@
+using Microsoft.EntityFrameworkCore;
+using Microsoft.Extensions.DependencyInjection;
+using Orleans.Persistence;
+using Orleans.Persistence.EntityFrameworkCore.SqlServer.Data;
+using Orleans.Reminders;
+using Orleans.Reminders.EntityFrameworkCore.SqlServer.Data;
+using Orleans.TestingHost;
+using TestExtensions;
+
+namespace Tester.EFCore;
+
+public class EFCoreFixture<TDbContext> : BaseTestClusterFixture where TDbContext : DbContext
+{
+    protected override void CheckPreconditionsOrThrow() => EFCoreTestUtils.CheckSqlServer();
+
+    protected override void ConfigureTestCluster(TestClusterBuilder builder)
+    {
+        builder.Options.InitialSilosCount = 4;
+        builder.AddSiloBuilderConfigurator<SiloConfigurator>();
+    }
+
+    private class SiloConfigurator : ISiloConfigurator
+    {
+        public void Configure(ISiloBuilder hostBuilder)
+        {
+            var ctxTypeName = typeof(TDbContext).Name;
+            var cs = $"Server=localhost,1433;Database=OrleansTests.{ctxTypeName};User Id=sa;Password=yourStrong(!)Password;TrustServerCertificate=True";
+
+            hostBuilder.Services.AddPooledDbContextFactory<TDbContext>(optionsBuilder =>
+            {
+                optionsBuilder.UseSqlServer(cs, opt =>
+                {
+                    opt.MigrationsHistoryTable("__EFMigrationsHistory");
+                    opt.MigrationsAssembly(typeof(TDbContext).Assembly.FullName);
+                });
+            });
+
+            switch (ctxTypeName)
+            {
+                case nameof(SqlServerGrainStateDbContext):
+                    hostBuilder
+                        .AddEntityFrameworkCoreSqlServerGrainStorage("GrainStorageForTest");
+                    break;
+                case nameof(SqlServerReminderDbContext):
+                    hostBuilder
+                        .UseEntityFrameworkCoreSqlServerReminderService();
+                    break;
+            }
+
+            hostBuilder
+                .AddMemoryGrainStorage("MemoryStore");
+
+            var sp = new ServiceCollection()
+                .AddPooledDbContextFactory<TDbContext>(optionsBuilder =>
+                {
+                    optionsBuilder.UseSqlServer(cs, opt =>
+                    {
+                        opt.MigrationsHistoryTable("__EFMigrationsHistory");
+                        opt.MigrationsAssembly(typeof(TDbContext).Assembly.FullName);
+                    });
+                }).BuildServiceProvider();
+
+            var factory = sp.GetRequiredService<IDbContextFactory<TDbContext>>();
+
+            var ctx = factory.CreateDbContext();
+            if (ctx.Database.GetPendingMigrations().Any())
+            {
+                try
+                {
+                    ctx.Database.Migrate();
+                }
+                catch { }
+            }
+        }
+    }
+}