Merge pull request #961 from omgbeez/upstream/race-fix

fix: Race conditions when deleting torrent and adding downloads.

server/RdtClient.Data/Data/DataContext.cs

@@ -16,6 +16,14 @@ public class DataContext(DbContextOptions options) : IdentityDbContext(options)
     {
         base.OnModelCreating(builder);
 
+        builder.Entity<Download>()
+               .HasIndex(m => new
+               {
+                   m.TorrentId,
+                   m.Path
+               })
+               .IsUnique();
+
         var cascadeFKs = builder.Model.GetEntityTypes()
                                 .SelectMany(t => t.GetForeignKeys())
                                 .Where(fk => !fk.IsOwnership && fk.DeleteBehavior == DeleteBehavior.Cascade);

server/RdtClient.Data/Data/DownloadAddResult.cs

@@ -0,0 +1,9 @@
+namespace RdtClient.Data.Data;
+
+public enum DownloadAddResult
+{
+    Added,
+    AlreadyExists,
+    TorrentMissing,
+    InvalidInput
+}

server/RdtClient.Data/Data/DownloadData.cs

@@ -1,10 +1,12 @@
-using Microsoft.EntityFrameworkCore;
+using Microsoft.Data.Sqlite;
+using Microsoft.EntityFrameworkCore;
+using Microsoft.Extensions.Logging;
 using RdtClient.Data.Models.Data;
 using Download = RdtClient.Data.Models.Data.Download;
 
 namespace RdtClient.Data.Data;
 
-public class DownloadData(DataContext dataContext)
+public class DownloadData(DataContext dataContext, ILogger<DownloadData>? logger = null)
 {
     public async Task<List<Download>> GetForTorrent(Guid torrentId)
     {
@@ -30,8 +32,29 @@ public class DownloadData(DataContext dataContext)
                                 .FirstOrDefaultAsync(m => m.TorrentId == torrentId && m.Path == path);
     }
 
-    public async Task<Download> Add(Guid torrentId, DownloadInfo downloadInfo)
+    public async Task<DownloadAddResult> TryAddForTorrent(Guid torrentId, DownloadInfo downloadInfo)
     {
+        if (String.IsNullOrWhiteSpace(downloadInfo.RestrictedLink))
+        {
+            logger?.LogDebug("Skipped download creation because the restricted link was blank. TorrentId: {torrentId}", torrentId);
+
+            return DownloadAddResult.InvalidInput;
+        }
+
+        if (!await dataContext.Torrents.AsNoTracking().AnyAsync(m => m.TorrentId == torrentId))
+        {
+            logger?.LogDebug("Skipped download creation because the torrent no longer exists. TorrentId: {torrentId}, Path: {path}", torrentId, downloadInfo.RestrictedLink);
+
+            return DownloadAddResult.TorrentMissing;
+        }
+
+        if (await dataContext.Downloads.AsNoTracking().AnyAsync(m => m.TorrentId == torrentId && m.Path == downloadInfo.RestrictedLink))
+        {
+            logger?.LogDebug("Skipped download creation because it already exists. TorrentId: {torrentId}, Path: {path}", torrentId, downloadInfo.RestrictedLink);
+
+            return DownloadAddResult.AlreadyExists;
+        }
+
         var download = new Download
         {
             DownloadId = Guid.NewGuid(),
@@ -45,9 +68,33 @@ public class DownloadData(DataContext dataContext)
 
         await dataContext.Downloads.AddAsync(download);
 
-        await dataContext.SaveChangesAsync();
+        try
+        {
+            await dataContext.SaveChangesAsync();
 
-        return download;
+            return DownloadAddResult.Added;
+        }
+        // These shouldn't be possible any longer, but added for safety and until confirmed.
+        catch (DbUpdateException ex)
+        {
+            dataContext.Entry(download).State = EntityState.Detached;
+
+            if (IsDuplicateDownloadViolation(ex))
+            {
+                logger?.LogDebug("Skipped download creation after a concurrent duplicate insert. TorrentId: {torrentId}, Path: {path}", torrentId, downloadInfo.RestrictedLink);
+
+                return DownloadAddResult.AlreadyExists;
+            }
+
+            if (IsForeignKeyViolation(ex) && !await dataContext.Torrents.AsNoTracking().AnyAsync(m => m.TorrentId == torrentId))
+            {
+                logger?.LogDebug("Skipped download creation after the torrent was deleted concurrently. TorrentId: {torrentId}, Path: {path}", torrentId, downloadInfo.RestrictedLink);
+
+                return DownloadAddResult.TorrentMissing;
+            }
+
+            throw;
+        }
     }
 
     public async Task UpdateUnrestrictedLink(Guid downloadId, String unrestrictedLink)
@@ -246,4 +293,20 @@ public class DownloadData(DataContext dataContext)
 
         await dataContext.SaveChangesAsync();
     }
+
+    private static Boolean IsDuplicateDownloadViolation(DbUpdateException exception)
+    {
+        var sqliteException = exception.InnerException as SqliteException;
+
+        return sqliteException?.SqliteExtendedErrorCode == 2067
+               || sqliteException?.Message.Contains("UNIQUE constraint failed: Downloads.TorrentId, Downloads.Path", StringComparison.Ordinal) == true;
+    }
+
+    private static Boolean IsForeignKeyViolation(DbUpdateException exception)
+    {
+        var sqliteException = exception.InnerException as SqliteException;
+
+        return sqliteException?.SqliteExtendedErrorCode == 787
+               || sqliteException?.Message.Contains("FOREIGN KEY constraint failed", StringComparison.Ordinal) == true;
+    }
 }

server/RdtClient.Data/Data/TorrentData.cs

@@ -1,10 +1,11 @@
-using Microsoft.EntityFrameworkCore;
+using Microsoft.EntityFrameworkCore;
+using Microsoft.Extensions.Logging;
 using RdtClient.Data.Enums;
 using RdtClient.Data.Models.Data;
 
 namespace RdtClient.Data.Data;
 
-public class TorrentData(DataContext dataContext) : ITorrentData
+public class TorrentData(DataContext dataContext, ILogger<TorrentData>? logger = null) : ITorrentData
 {
     public async Task<IList<Torrent>> Get()
     {
@@ -286,15 +287,23 @@ public class TorrentData(DataContext dataContext) : ITorrentData
 
     public async Task Delete(Guid torrentId)
     {
-        var dbTorrent = await dataContext.Torrents.FirstOrDefaultAsync(m => m.TorrentId == torrentId);
+        await using var transaction = await dataContext.Database.BeginTransactionAsync();
 
-        if (dbTorrent == null)
+        await dataContext.Downloads
+                         .Where(m => m.TorrentId == torrentId)
+                         .ExecuteDeleteAsync();
+
+        var deletedTorrents = await dataContext.Torrents
+                                               .Where(m => m.TorrentId == torrentId)
+                                               .ExecuteDeleteAsync();
+
+        await transaction.CommitAsync();
+
+        if (deletedTorrents == 0)
         {
+            logger?.LogDebug("Skipped torrent graph deletion because the torrent was not found. TorrentId: {torrentId}", torrentId);
+
             return;
         }
-
-        dataContext.Torrents.Remove(dbTorrent);
-
-        await dataContext.SaveChangesAsync();
     }
 }

server/RdtClient.Data/Migrations/20260423031719_Downloads_Add_TorrentIdPath_Unique.cs

@@ -0,0 +1,48 @@
+using Microsoft.EntityFrameworkCore.Infrastructure;
+using Microsoft.EntityFrameworkCore.Migrations;
+using RdtClient.Data.Data;
+
+#nullable disable
+
+namespace RdtClient.Data.Migrations;
+
+[DbContext(typeof(DataContext))]
+[Migration("20260423031719_Downloads_Add_TorrentIdPath_Unique")]
+public partial class Downloads_Add_TorrentIdPath_Unique : Migration
+{
+    protected override void Up(MigrationBuilder migrationBuilder)
+    {
+        // Orphaned downloads from previous failures.
+        migrationBuilder.Sql("""
+                             DELETE FROM Downloads
+                             WHERE rowid NOT IN (
+                                 SELECT MIN(rowid)
+                                 FROM Downloads
+                                 GROUP BY TorrentId, Path
+                             );
+                             """);
+
+        migrationBuilder.DropIndex(
+            name: "IX_Downloads_TorrentId",
+            table: "Downloads");
+
+        // Prevent accidental duplicates, provides idempotency at the DB level for downloads
+        migrationBuilder.CreateIndex(
+            name: "IX_Downloads_TorrentId_Path",
+            table: "Downloads",
+            columns: ["TorrentId", "Path"],
+            unique: true);
+    }
+
+    protected override void Down(MigrationBuilder migrationBuilder)
+    {
+        migrationBuilder.DropIndex(
+            name: "IX_Downloads_TorrentId_Path",
+            table: "Downloads");
+
+        migrationBuilder.CreateIndex(
+            name: "IX_Downloads_TorrentId",
+            table: "Downloads",
+            column: "TorrentId");
+    }
+}

server/RdtClient.Data/Migrations/DataContextModelSnapshot.cs

@@ -15,7 +15,7 @@ namespace RdtClient.Data.Migrations
         protected override void BuildModel(ModelBuilder modelBuilder)
         {
 #pragma warning disable 612, 618
-            modelBuilder.HasAnnotation("ProductVersion", "9.0.9");
+            modelBuilder.HasAnnotation("ProductVersion", "10.0.5");
 
             modelBuilder.Entity("Microsoft.AspNetCore.Identity.IdentityRole", b =>
                 {
@@ -263,7 +263,8 @@ namespace RdtClient.Data.Migrations
 
                     b.HasKey("DownloadId");
 
-                    b.HasIndex("TorrentId");
+                    b.HasIndex("TorrentId", "Path")
+                        .IsUnique();
 
                     b.ToTable("Downloads");
                 });

server/RdtClient.Service.Test/Regression/TorrentDownloadRaceTests.cs

@@ -0,0 +1,183 @@
+using Microsoft.Data.Sqlite;
+using Microsoft.EntityFrameworkCore;
+using RdtClient.Data.Data;
+using RdtClient.Data.Models.Data;
+
+namespace RdtClient.Service.Test.Regression;
+
+public class TorrentDownloadRaceTests : IAsyncLifetime
+{
+    private readonly String _databasePath = Path.Combine(Path.GetTempPath(), $"rdt-client-race-{Guid.NewGuid():N}.sqlite");
+
+    [Fact]
+    public async Task Add_WhenTorrentIsDeletedAfterItWasRead_DoesNotThrowAndDoesNotInsertDownload()
+    {
+        var torrentId = await SeedTorrentAsync();
+
+        await using (var readerContext = CreateContext())
+        {
+            var torrentData = new TorrentData(readerContext);
+            var torrent = await torrentData.GetById(torrentId);
+
+            Assert.NotNull(torrent);
+        }
+
+        await using (var deleteContext = CreateContext())
+        {
+            var torrentData = new TorrentData(deleteContext);
+            await torrentData.Delete(torrentId);
+        }
+
+        DownloadAddResult result;
+
+        await using (var addContext = CreateContext())
+        {
+            var downloadData = new DownloadData(addContext);
+            result = await downloadData.TryAddForTorrent(torrentId, CreateDownloadInfo("race-a"));
+        }
+
+        Assert.Equal(DownloadAddResult.TorrentMissing, result);
+
+        await using var verifyContext = CreateContext();
+        Assert.False(await verifyContext.Torrents.AnyAsync(m => m.TorrentId == torrentId));
+        Assert.False(await verifyContext.Downloads.AnyAsync());
+    }
+
+    [Fact]
+    public async Task Delete_WhenDownloadIsInsertedBetweenChildAndParentDelete_DoesNotThrowAndRemovesTorrentGraph()
+    {
+        var torrentId = await SeedTorrentAsync();
+
+        await using (var deleteChildrenContext = CreateContext())
+        {
+            var downloadData = new DownloadData(deleteChildrenContext);
+            await downloadData.DeleteForTorrent(torrentId);
+        }
+
+        await using (var addContext = CreateContext())
+        {
+            var downloadData = new DownloadData(addContext);
+            var result = await downloadData.TryAddForTorrent(torrentId, CreateDownloadInfo("race-b"));
+            Assert.Equal(DownloadAddResult.Added, result);
+        }
+
+        Exception? exception;
+
+        await using (var deleteParentContext = CreateContext())
+        {
+            var torrentData = new TorrentData(deleteParentContext);
+            exception = await Record.ExceptionAsync(() => torrentData.Delete(torrentId));
+        }
+
+        Assert.Null(exception);
+
+        await using var verifyContext = CreateContext();
+        Assert.False(await verifyContext.Torrents.AnyAsync(m => m.TorrentId == torrentId));
+        Assert.False(await verifyContext.Downloads.AnyAsync(m => m.TorrentId == torrentId));
+    }
+
+    [Fact]
+    public async Task TryAddForTorrent_WhenDownloadAlreadyExists_DoesNotInsertDuplicate()
+    {
+        var torrentId = await SeedTorrentAsync();
+
+        await using (var firstContext = CreateContext())
+        {
+            var downloadData = new DownloadData(firstContext);
+            var result = await downloadData.TryAddForTorrent(torrentId, CreateDownloadInfo("race-c"));
+
+            Assert.Equal(DownloadAddResult.Added, result);
+        }
+
+        await using (var secondContext = CreateContext())
+        {
+            var downloadData = new DownloadData(secondContext);
+            var result = await downloadData.TryAddForTorrent(torrentId, CreateDownloadInfo("race-c"));
+
+            Assert.Equal(DownloadAddResult.AlreadyExists, result);
+        }
+
+        await using var verifyContext = CreateContext();
+        Assert.Equal(1, await verifyContext.Downloads.CountAsync(m => m.TorrentId == torrentId));
+    }
+
+    [Fact]
+    public async Task Delete_WhenTorrentWasAlreadyDeleted_DoesNotThrow()
+    {
+        var torrentId = await SeedTorrentAsync();
+
+        await using (var firstDeleteContext = CreateContext())
+        {
+            var torrentData = new TorrentData(firstDeleteContext);
+            await torrentData.Delete(torrentId);
+        }
+
+        Exception? exception;
+
+        await using (var secondDeleteContext = CreateContext())
+        {
+            var torrentData = new TorrentData(secondDeleteContext);
+            exception = await Record.ExceptionAsync(() => torrentData.Delete(torrentId));
+        }
+
+        Assert.Null(exception);
+    }
+
+    public async Task InitializeAsync()
+    {
+        await using var context = CreateContext();
+        await context.Database.EnsureDeletedAsync();
+        await context.Database.EnsureCreatedAsync();
+    }
+
+    public Task DisposeAsync()
+    {
+        if (File.Exists(_databasePath))
+        {
+            File.Delete(_databasePath);
+        }
+
+        return Task.CompletedTask;
+    }
+
+    private DataContext CreateContext()
+    {
+        var connectionString = new SqliteConnectionStringBuilder
+        {
+            DataSource = _databasePath,
+            ForeignKeys = true
+        }.ToString();
+
+        var options = new DbContextOptionsBuilder<DataContext>()
+                      .UseSqlite(connectionString)
+                      .Options;
+
+        return new(options);
+    }
+
+    private async Task<Guid> SeedTorrentAsync()
+    {
+        var torrentId = Guid.NewGuid();
+
+        await using var context = CreateContext();
+        context.Torrents.Add(new Torrent
+        {
+            TorrentId = torrentId,
+            Hash = Guid.NewGuid().ToString("N"),
+            Added = DateTimeOffset.UtcNow
+        });
+
+        await context.SaveChangesAsync();
+
+        return torrentId;
+    }
+
+    private static DownloadInfo CreateDownloadInfo(String suffix)
+    {
+        return new()
+        {
+            FileName = $"download-{suffix}.bin",
+            RestrictedLink = $"https://example.invalid/{suffix}"
+        };
+    }
+}

server/RdtClient.Service.Test/Services/TorrentClients/TorBoxDebridClientTest.cs

@@ -911,4 +911,30 @@ public class TorBoxDebridClientTest
                                       It.Is<Func<It.IsAnyType, Exception?, String>>((v, t) => true)),
                            Times.Once);
     }
+
+    [Fact]
+    public async Task UpdateData_SetsProcessingStatus_WhenTorBoxStatusIsUnmappedAndTorrentWasQueued()
+    {
+        // Arrange
+        var torrent = new Torrent
+        {
+            RdId = "test-rd-id",
+            RdStatus = TorrentStatus.Queued,
+            RdName = "test-torrent"
+        };
+
+        var torrentClientTorrent = new DebridClientTorrent
+        {
+            Status = "some-unknown-status",
+            Filename = "test-torrent"
+        };
+
+        var clientMock = new Mock<TorBoxDebridClient>(_loggerMock.Object, _httpClientFactoryMock.Object, _fileFilterMock.Object, _coordinatorMock.Object);
+
+        // Act
+        var result = await clientMock.Object.UpdateData(torrent, torrentClientTorrent);
+
+        // Assert
+        Assert.Equal(TorrentStatus.Processing, result.RdStatus);
+    }
 }

server/RdtClient.Service/BackgroundServices/ProviderUpdater.cs

@@ -18,13 +18,14 @@ public class ProviderUpdater(ILogger<ProviderUpdater> logger, IServiceProvider s
             await Task.Delay(1000, stoppingToken);
         }
 
-        using var scope = serviceProvider.CreateScope();
-        var torrentService = scope.ServiceProvider.GetRequiredService<Torrents>();
-        var torrentRunner = scope.ServiceProvider.GetRequiredService<TorrentRunner>();
         logger.LogInformation("ProviderUpdater started.");
 
         while (!stoppingToken.IsCancellationRequested)
         {
+            using var scope = serviceProvider.CreateScope();
+            var torrentService = scope.ServiceProvider.GetRequiredService<Torrents>();
+            var torrentRunner = scope.ServiceProvider.GetRequiredService<TorrentRunner>();
+
             try
             {
                 var torrents = await torrentService.Get();

server/RdtClient.Service/BackgroundServices/TaskRunner.cs

@@ -16,15 +16,19 @@ public class TaskRunner(ILogger<TaskRunner> logger, IServiceProvider serviceProv
             await Task.Delay(1000, stoppingToken);
         }
 
-        using var scope = serviceProvider.CreateScope();
-        var torrentRunner = scope.ServiceProvider.GetRequiredService<TorrentRunner>();
-
         logger.LogInformation("TaskRunner started.");
 
-        await torrentRunner.Initialize();
+        using (var startupScope = serviceProvider.CreateScope())
+        {
+            var startupRunner = startupScope.ServiceProvider.GetRequiredService<TorrentRunner>();
+            await startupRunner.Initialize();
+        }
 
         while (!stoppingToken.IsCancellationRequested)
         {
+            using var scope = serviceProvider.CreateScope();
+            var torrentRunner = scope.ServiceProvider.GetRequiredService<TorrentRunner>();
+
             try
             {
                 await torrentRunner.Tick();

server/RdtClient.Service/Services/DebridClients/TorBoxDebridClient.cs

@@ -644,10 +644,16 @@ public class TorBoxDebridClient(ILogger<TorBoxDebridClient> logger, IHttpClientF
     {
         if (!String.IsNullOrWhiteSpace(status))
         {
-            logger.LogInformation("TorBoxDebridClient encountered an unmapped status: {Status} for torrent {TorrentName}", status, torrent.RdName);
+            logger.LogInformation("TorBoxDebridClient encountered an unmapped status: {Status} for torrent {TorrentName} with previous status {PreviousStatus}",
+                                  status,
+                                  torrent.RdName,
+                                  torrent.RdStatus);
         }
 
-        return torrent.RdStatus ?? TorrentStatus.Processing;
+        // Once TorBox has acknowledged the torrent, an unknown provider-side status should not fall all the way back to
+        // the local queue state. Treat it as provider-side processing so the UI does not incorrectly show
+        // "Not Yet Added to Provider" while TorBox is already handling it.
+        return torrent.RdStatus is null or TorrentStatus.Queued ? TorrentStatus.Processing : torrent.RdStatus.Value;
     }
 
     private void Log(String message, Torrent? torrent = null)

server/RdtClient.Service/Services/Downloads.cs

@@ -21,9 +21,9 @@ public class Downloads(DownloadData downloadData) : IDownloads
         return await downloadData.Get(torrentId, path);
     }
 
-    public async Task<Download> Add(Guid torrentId, DownloadInfo downloadInfo)
+    public async Task<DownloadAddResult> TryAddForTorrent(Guid torrentId, DownloadInfo downloadInfo)
     {
-        return await downloadData.Add(torrentId, downloadInfo);
+        return await downloadData.TryAddForTorrent(torrentId, downloadInfo);
     }
 
     public async Task UpdateUnrestrictedLink(Guid downloadId, String unrestrictedLink)

server/RdtClient.Service/Services/IDownloads.cs

@@ -1,3 +1,4 @@
+using RdtClient.Data.Data;
 using RdtClient.Data.Models.Data;
 
 namespace RdtClient.Service.Services;
@@ -7,7 +8,7 @@ public interface IDownloads
     Task<List<Download>> GetForTorrent(Guid torrentId);
     Task<Download?> GetById(Guid downloadId);
     Task<Download?> Get(Guid torrentId, String path);
-    Task<Download> Add(Guid torrentId, DownloadInfo downloadInfo);
+    Task<DownloadAddResult> TryAddForTorrent(Guid torrentId, DownloadInfo downloadInfo);
     Task UpdateUnrestrictedLink(Guid downloadId, String unrestrictedLink);
     Task UpdateFileName(Guid downloadId, String fileName);
     Task UpdateDownloadStarted(Guid downloadId, DateTimeOffset? dateTime);

server/RdtClient.Service/Services/Torrents.cs

@@ -457,12 +457,23 @@ public class Torrents(
 
         foreach (var downloadInfo in downloadInfos)
         {
-            // Make sure downloads don't get added multiple times
-            var downloadExists = await downloads.Get(torrent.TorrentId, downloadInfo.RestrictedLink);
+            var addResult = await downloads.TryAddForTorrent(torrent.TorrentId, downloadInfo);
 
-            if (downloadExists == null && !String.IsNullOrWhiteSpace(downloadInfo.RestrictedLink))
+            switch (addResult)
             {
-                await downloads.Add(torrent.TorrentId, downloadInfo);
+                case DownloadAddResult.Added:
+                case DownloadAddResult.AlreadyExists:
+                    continue;
+                case DownloadAddResult.TorrentMissing:
+                    logger.LogDebug("Stopping download creation because the torrent was deleted concurrently. TorrentId: {torrentId}", torrent.TorrentId);
+
+                    return;
+                case DownloadAddResult.InvalidInput:
+                    logger.LogDebug("Skipping download creation because the provider returned an invalid download link. TorrentId: {torrentId}", torrent.TorrentId);
+
+                    continue;
+                default:
+                    throw new ArgumentOutOfRangeException(nameof(addResult), addResult, null);
             }
         }
     }
@@ -539,7 +550,6 @@ public class Torrents(
         {
             Log($"Deleting RdtClient data", torrent);
 
-            await downloads.DeleteForTorrent(torrent.TorrentId);
             await torrentData.Delete(torrentId);
         }
 
@@ -1070,7 +1080,7 @@ public class Torrents(
                 await torrentData.UpdateRdData(torrent);
             }
         }
-        catch
+        catch (Exception ex)
         {
             // ignored
         }