diff --git a/ErsatzTV.Application/ConcurrencyExtensions.cs b/ErsatzTV.Application/ConcurrencyExtensions.cs index e98cdabd9..5d29c7d31 100644 --- a/ErsatzTV.Application/ConcurrencyExtensions.cs +++ b/ErsatzTV.Application/ConcurrencyExtensions.cs @@ -1,6 +1,7 @@ using ErsatzTV.Core; using ErsatzTV.Core.Domain; using ErsatzTV.Core.Errors; +using ErsatzTV.Infrastructure.Data; using Microsoft.EntityFrameworkCore; using Microsoft.EntityFrameworkCore.ChangeTracking; @@ -70,6 +71,32 @@ public static class ConcurrencyExtensions } } + /// + /// Like , but additionally treats a unique / primary-key + /// constraint violation as an idempotent no-op: returns false instead of throwing when the + /// save fails because a concurrent request inserted a row we had membership-checked absent (the + /// composite-PK race on CollectionItem — issue #308). A false means "the desired row + /// already exists because a racing writer won; the winner ran the ETag rotation + fan-out, so skip + /// ours." true means our own change committed. Every other + /// (and the genuine deleted-row concurrency conflict rethrown by ) + /// still propagates. The only insert these callers stage is the CollectionItem join row, so the + /// sole unique/PK constraint that can fire here is that composite key. + /// + public static async Task TrySaveChangesForcingVersion( + this DbContext dbContext, + CancellationToken cancellationToken) + { + try + { + await dbContext.SaveChangesForcingVersion(cancellationToken); + return true; + } + catch (DbUpdateException ex) when (TvContext.IsUniqueConstraintViolation(ex)) + { + return false; + } + } + /// /// Persist pending changes, mapping the EF optimistic-concurrency failure to /// (→ 412). When a versioned root carries an diff --git a/ErsatzTV.Application/MediaCollections/Commands/AddArtistToCollectionHandler.cs b/ErsatzTV.Application/MediaCollections/Commands/AddArtistToCollectionHandler.cs index df3c4c829..2a15b8349 100644 --- a/ErsatzTV.Application/MediaCollections/Commands/AddArtistToCollectionHandler.cs +++ b/ErsatzTV.Application/MediaCollections/Commands/AddArtistToCollectionHandler.cs @@ -1,4 +1,4 @@ -using System.Threading.Channels; +using System.Threading.Channels; using ErsatzTV.Application.Playouts; using ErsatzTV.Application.Search; using ErsatzTV.Core; @@ -55,7 +55,13 @@ public class AddArtistToCollectionHandler : // force-write past a concurrent replace-all Version bump — this add takes no If-Match, so a // benign race must not 500 (#253/#269 §7a). Post-commit enqueues run on CancellationToken.None. parameters.Collection.Version++; - await dbContext.SaveChangesForcingVersion(CancellationToken.None); + if (!await dbContext.TrySaveChangesForcingVersion(CancellationToken.None)) + { + // A concurrent add of this same item won the composite-PK race and already inserted the row, + // rotated the collection ETag, and fanned out the rebuild — so this is now an idempotent + // no-op. Skip our reindex/rebuild fan-out (the winner already did it). #308 + return Unit.Default; + } await _searchChannel.WriteAsync(new ReindexMediaItems([parameters.Artist.Id]), CancellationToken.None); diff --git a/ErsatzTV.Application/MediaCollections/Commands/AddEpisodeToCollectionHandler.cs b/ErsatzTV.Application/MediaCollections/Commands/AddEpisodeToCollectionHandler.cs index b46405fc7..223c94fa7 100644 --- a/ErsatzTV.Application/MediaCollections/Commands/AddEpisodeToCollectionHandler.cs +++ b/ErsatzTV.Application/MediaCollections/Commands/AddEpisodeToCollectionHandler.cs @@ -1,4 +1,4 @@ -using System.Threading.Channels; +using System.Threading.Channels; using ErsatzTV.Application.Playouts; using ErsatzTV.Application.Search; using ErsatzTV.Core; @@ -57,7 +57,13 @@ public class AddEpisodeToCollectionHandler : // force-write past a concurrent replace-all Version bump — this add takes no If-Match, so a // benign race must not 500 (#253/#269 §7a). Post-commit enqueues run on CancellationToken.None. parameters.Collection.Version++; - await dbContext.SaveChangesForcingVersion(CancellationToken.None); + if (!await dbContext.TrySaveChangesForcingVersion(CancellationToken.None)) + { + // A concurrent add of this same item won the composite-PK race and already inserted the row, + // rotated the collection ETag, and fanned out the rebuild — so this is now an idempotent + // no-op. Skip our reindex/rebuild fan-out (the winner already did it). #308 + return Unit.Default; + } await _searchChannel.WriteAsync(new ReindexMediaItems([parameters.Episode.Id]), CancellationToken.None); diff --git a/ErsatzTV.Application/MediaCollections/Commands/AddImageToCollectionHandler.cs b/ErsatzTV.Application/MediaCollections/Commands/AddImageToCollectionHandler.cs index 2eba28437..6a6b81656 100644 --- a/ErsatzTV.Application/MediaCollections/Commands/AddImageToCollectionHandler.cs +++ b/ErsatzTV.Application/MediaCollections/Commands/AddImageToCollectionHandler.cs @@ -1,4 +1,4 @@ -using System.Threading.Channels; +using System.Threading.Channels; using ErsatzTV.Application.Playouts; using ErsatzTV.Application.Search; using ErsatzTV.Core; @@ -54,7 +54,13 @@ public class AddImageToCollectionHandler : IRequestHandler> Handle( AddItemsToCollection request, CancellationToken cancellationToken) { - await using TvContext dbContext = await _dbContextFactory.CreateDbContextAsync(cancellationToken); - Option maybeCollection = await CollectionMustExist(dbContext, request, cancellationToken); - return await maybeCollection.Match( - Some: async collection => + for (var attempt = 0; ; attempt++) + { + await using TvContext dbContext = await _dbContextFactory.CreateDbContextAsync(cancellationToken); + Option maybeCollection = await CollectionMustExist(dbContext, request, cancellationToken); + + // true = terminal (nothing to add, or our batch committed); false = a duplicate-key race + // rolled the batch back, recompute membership and retry. + Either attemptResult = await maybeCollection.Match( + Some: async collection => + { + Validation validation = await Validate(dbContext, request, collection, cancellationToken); + return await validation.Apply(c => ApplyAddItemsRequest(dbContext, c, request, cancellationToken)); + }, + None: () => Task.FromResult>( + new NotFoundError($"Collection {request.CollectionId} does not exist."))); + + if (attemptResult.IsLeft) { - Validation validation = await Validate(dbContext, request, collection, cancellationToken); - return await validation.Apply(c => ApplyAddItemsRequest(dbContext, c, request, cancellationToken)); - }, - None: () => Task.FromResult>( - new NotFoundError($"Collection {request.CollectionId} does not exist."))); + return attemptResult.Map(_ => Unit.Default); + } + + bool committed = attemptResult.Match(Left: _ => false, Right: done => done); + if (committed) + { + return Unit.Default; + } + + // A concurrent add inserted one+ of our items first; recompute against fresh membership. + if (attempt >= MaxDuplicateRetries) + { + return BaseError.New( + "Concurrent modification while adding items to the collection; please retry."); + } + } } - private async Task ApplyAddItemsRequest( + private async Task ApplyAddItemsRequest( TvContext dbContext, Collection collection, AddItemsToCollection request, @@ -75,10 +104,10 @@ public class AddItemsToCollectionHandler : var toAddIds = allItems.Where(item => collection.MediaItems.All(mi => mi.Id != item)).ToList(); // No-op when every requested item is already a member: don't rotate the ETag or fan out - // rebuilds for an idempotent re-add — #269. + // rebuilds for an idempotent re-add — #269. Terminal success (no retry). if (toAddIds.Count == 0) { - return Unit.Default; + return true; } List toAdd = await dbContext.MediaItems @@ -91,7 +120,15 @@ public class AddItemsToCollectionHandler : // force-write past a concurrent replace-all Version bump — this add takes no If-Match, so a // benign race must not 500 (#253/#269 §7a). collection.Version++; - await dbContext.SaveChangesForcingVersion(cancellationToken); + + // A concurrent add of an overlapping item won the composite-PK race and rolled back this whole + // batch. Unlike the single-item handlers (idempotent no-op), a bulk add must NOT drop the items + // that did NOT collide — signal the caller to recompute membership and retry the still-missing + // ones. #308 + if (!await dbContext.TrySaveChangesForcingVersion(cancellationToken)) + { + return false; + } // post-commit side effect runs on CancellationToken.None so a late request cancellation // can't abort it after the commit landed (#254) @@ -104,7 +141,7 @@ public class AddItemsToCollectionHandler : await _channel.WriteAsync(new BuildPlayout(playoutId, PlayoutBuildMode.Refresh), CancellationToken.None); } - return Unit.Default; + return true; } private async Task> Validate( diff --git a/ErsatzTV.Application/MediaCollections/Commands/AddMediaItemToCollectionHandler.cs b/ErsatzTV.Application/MediaCollections/Commands/AddMediaItemToCollectionHandler.cs index 195e6cff9..9949d124d 100644 --- a/ErsatzTV.Application/MediaCollections/Commands/AddMediaItemToCollectionHandler.cs +++ b/ErsatzTV.Application/MediaCollections/Commands/AddMediaItemToCollectionHandler.cs @@ -1,4 +1,4 @@ -using System.Threading.Channels; +using System.Threading.Channels; using ErsatzTV.Application.Playouts; using ErsatzTV.Application.Search; using ErsatzTV.Core; @@ -55,7 +55,13 @@ public class AddMediaItemToCollectionHandler : // force-write past a concurrent replace-all Version bump — this add takes no If-Match, so a // benign race must not 500 (#253/#269 §7a). Post-commit enqueues run on CancellationToken.None. parameters.Collection.Version++; - await dbContext.SaveChangesForcingVersion(CancellationToken.None); + if (!await dbContext.TrySaveChangesForcingVersion(CancellationToken.None)) + { + // A concurrent add of this same item won the composite-PK race and already inserted the row, + // rotated the collection ETag, and fanned out the rebuild — so this is now an idempotent + // no-op. Skip our reindex/rebuild fan-out (the winner already did it). #308 + return Unit.Default; + } await _searchChannel.WriteAsync(new ReindexMediaItems([parameters.MediaItem.Id]), CancellationToken.None); diff --git a/ErsatzTV.Application/MediaCollections/Commands/AddMovieToCollectionHandler.cs b/ErsatzTV.Application/MediaCollections/Commands/AddMovieToCollectionHandler.cs index 603d4d49a..54290928e 100644 --- a/ErsatzTV.Application/MediaCollections/Commands/AddMovieToCollectionHandler.cs +++ b/ErsatzTV.Application/MediaCollections/Commands/AddMovieToCollectionHandler.cs @@ -1,4 +1,4 @@ -using System.Threading.Channels; +using System.Threading.Channels; using ErsatzTV.Application.Playouts; using ErsatzTV.Application.Search; using ErsatzTV.Core; @@ -55,7 +55,13 @@ public class AddMovieToCollectionHandler : // force-write past a concurrent replace-all Version bump — this add takes no If-Match, so a // benign race must not 500 (#253/#269 §7a). Post-commit enqueues run on CancellationToken.None. parameters.Collection.Version++; - await dbContext.SaveChangesForcingVersion(CancellationToken.None); + if (!await dbContext.TrySaveChangesForcingVersion(CancellationToken.None)) + { + // A concurrent add of this same item won the composite-PK race and already inserted the row, + // rotated the collection ETag, and fanned out the rebuild — so this is now an idempotent + // no-op. Skip our reindex/rebuild fan-out (the winner already did it). #308 + return Unit.Default; + } await _searchChannel.WriteAsync(new ReindexMediaItems([parameters.Movie.Id]), CancellationToken.None); diff --git a/ErsatzTV.Application/MediaCollections/Commands/AddMusicVideoToCollectionHandler.cs b/ErsatzTV.Application/MediaCollections/Commands/AddMusicVideoToCollectionHandler.cs index 39b2b21f4..29bc706fe 100644 --- a/ErsatzTV.Application/MediaCollections/Commands/AddMusicVideoToCollectionHandler.cs +++ b/ErsatzTV.Application/MediaCollections/Commands/AddMusicVideoToCollectionHandler.cs @@ -1,4 +1,4 @@ -using System.Threading.Channels; +using System.Threading.Channels; using ErsatzTV.Application.Playouts; using ErsatzTV.Application.Search; using ErsatzTV.Core; @@ -57,7 +57,13 @@ public class AddMusicVideoToCollectionHandler : // force-write past a concurrent replace-all Version bump — this add takes no If-Match, so a // benign race must not 500 (#253/#269 §7a). Post-commit enqueues run on CancellationToken.None. parameters.Collection.Version++; - await dbContext.SaveChangesForcingVersion(CancellationToken.None); + if (!await dbContext.TrySaveChangesForcingVersion(CancellationToken.None)) + { + // A concurrent add of this same item won the composite-PK race and already inserted the row, + // rotated the collection ETag, and fanned out the rebuild — so this is now an idempotent + // no-op. Skip our reindex/rebuild fan-out (the winner already did it). #308 + return Unit.Default; + } await _searchChannel.WriteAsync(new ReindexMediaItems([parameters.MusicVideo.Id]), CancellationToken.None); diff --git a/ErsatzTV.Application/MediaCollections/Commands/AddOtherVideoToCollectionHandler.cs b/ErsatzTV.Application/MediaCollections/Commands/AddOtherVideoToCollectionHandler.cs index c76fcfa48..d54cdddab 100644 --- a/ErsatzTV.Application/MediaCollections/Commands/AddOtherVideoToCollectionHandler.cs +++ b/ErsatzTV.Application/MediaCollections/Commands/AddOtherVideoToCollectionHandler.cs @@ -1,4 +1,4 @@ -using System.Threading.Channels; +using System.Threading.Channels; using ErsatzTV.Application.Playouts; using ErsatzTV.Application.Search; using ErsatzTV.Core; @@ -57,7 +57,13 @@ public class AddOtherVideoToCollectionHandler : // force-write past a concurrent replace-all Version bump — this add takes no If-Match, so a // benign race must not 500 (#253/#269 §7a). Post-commit enqueues run on CancellationToken.None. parameters.Collection.Version++; - await dbContext.SaveChangesForcingVersion(CancellationToken.None); + if (!await dbContext.TrySaveChangesForcingVersion(CancellationToken.None)) + { + // A concurrent add of this same item won the composite-PK race and already inserted the row, + // rotated the collection ETag, and fanned out the rebuild — so this is now an idempotent + // no-op. Skip our reindex/rebuild fan-out (the winner already did it). #308 + return Unit.Default; + } await _searchChannel.WriteAsync(new ReindexMediaItems([parameters.OtherVideo.Id]), CancellationToken.None); diff --git a/ErsatzTV.Application/MediaCollections/Commands/AddSeasonToCollectionHandler.cs b/ErsatzTV.Application/MediaCollections/Commands/AddSeasonToCollectionHandler.cs index 4613010eb..b3386a9bc 100644 --- a/ErsatzTV.Application/MediaCollections/Commands/AddSeasonToCollectionHandler.cs +++ b/ErsatzTV.Application/MediaCollections/Commands/AddSeasonToCollectionHandler.cs @@ -1,4 +1,4 @@ -using System.Threading.Channels; +using System.Threading.Channels; using ErsatzTV.Application.Playouts; using ErsatzTV.Application.Search; using ErsatzTV.Core; @@ -55,7 +55,13 @@ public class AddSeasonToCollectionHandler : // force-write past a concurrent replace-all Version bump — this add takes no If-Match, so a // benign race must not 500 (#253/#269 §7a). Post-commit enqueues run on CancellationToken.None. parameters.Collection.Version++; - await dbContext.SaveChangesForcingVersion(CancellationToken.None); + if (!await dbContext.TrySaveChangesForcingVersion(CancellationToken.None)) + { + // A concurrent add of this same item won the composite-PK race and already inserted the row, + // rotated the collection ETag, and fanned out the rebuild — so this is now an idempotent + // no-op. Skip our reindex/rebuild fan-out (the winner already did it). #308 + return Unit.Default; + } await _searchChannel.WriteAsync(new ReindexMediaItems([parameters.Season.Id]), CancellationToken.None); diff --git a/ErsatzTV.Application/MediaCollections/Commands/AddShowToCollectionHandler.cs b/ErsatzTV.Application/MediaCollections/Commands/AddShowToCollectionHandler.cs index f5bbc9fa0..e1fedb8dd 100644 --- a/ErsatzTV.Application/MediaCollections/Commands/AddShowToCollectionHandler.cs +++ b/ErsatzTV.Application/MediaCollections/Commands/AddShowToCollectionHandler.cs @@ -1,4 +1,4 @@ -using System.Threading.Channels; +using System.Threading.Channels; using ErsatzTV.Application.Playouts; using ErsatzTV.Application.Search; using ErsatzTV.Core; @@ -55,7 +55,13 @@ public class AddShowToCollectionHandler : // force-write past a concurrent replace-all Version bump — this add takes no If-Match, so a // benign race must not 500 (#253/#269 §7a). Post-commit enqueues run on CancellationToken.None. parameters.Collection.Version++; - await dbContext.SaveChangesForcingVersion(CancellationToken.None); + if (!await dbContext.TrySaveChangesForcingVersion(CancellationToken.None)) + { + // A concurrent add of this same item won the composite-PK race and already inserted the row, + // rotated the collection ETag, and fanned out the rebuild — so this is now an idempotent + // no-op. Skip our reindex/rebuild fan-out (the winner already did it). #308 + return Unit.Default; + } await _searchChannel.WriteAsync(new ReindexMediaItems([parameters.Show.Id]), CancellationToken.None); diff --git a/ErsatzTV.Application/MediaCollections/Commands/AddSongToCollectionHandler.cs b/ErsatzTV.Application/MediaCollections/Commands/AddSongToCollectionHandler.cs index aea054ed2..084681c99 100644 --- a/ErsatzTV.Application/MediaCollections/Commands/AddSongToCollectionHandler.cs +++ b/ErsatzTV.Application/MediaCollections/Commands/AddSongToCollectionHandler.cs @@ -1,4 +1,4 @@ -using System.Threading.Channels; +using System.Threading.Channels; using ErsatzTV.Application.Playouts; using ErsatzTV.Application.Search; using ErsatzTV.Core; @@ -55,7 +55,13 @@ public class AddSongToCollectionHandler : // force-write past a concurrent replace-all Version bump — this add takes no If-Match, so a // benign race must not 500 (#253/#269 §7a). Post-commit enqueues run on CancellationToken.None. parameters.Collection.Version++; - await dbContext.SaveChangesForcingVersion(CancellationToken.None); + if (!await dbContext.TrySaveChangesForcingVersion(CancellationToken.None)) + { + // A concurrent add of this same item won the composite-PK race and already inserted the row, + // rotated the collection ETag, and fanned out the rebuild — so this is now an idempotent + // no-op. Skip our reindex/rebuild fan-out (the winner already did it). #308 + return Unit.Default; + } await _searchChannel.WriteAsync(new ReindexMediaItems([parameters.Song.Id]), CancellationToken.None); diff --git a/ErsatzTV.Infrastructure.MySql/Data/MySqlErrorClassifier.cs b/ErsatzTV.Infrastructure.MySql/Data/MySqlErrorClassifier.cs new file mode 100644 index 000000000..c22d85590 --- /dev/null +++ b/ErsatzTV.Infrastructure.MySql/Data/MySqlErrorClassifier.cs @@ -0,0 +1,17 @@ +using Microsoft.EntityFrameworkCore; +using MySqlConnector; + +namespace ErsatzTV.Infrastructure.MySql.Data; + +/// +/// Classifies MySQL provider exceptions wrapped by EF Core. Wired to +/// at startup. +/// +public static class MySqlErrorClassifier +{ + // ER_DUP_ENTRY (1062): a duplicate value for a PRIMARY KEY or UNIQUE index. + private const int ErDupEntry = 1062; + + public static bool IsUniqueConstraintViolation(DbUpdateException ex) => + ex.InnerException is MySqlException { Number: ErDupEntry }; +} diff --git a/ErsatzTV.Infrastructure.Sqlite/Data/SqliteErrorClassifier.cs b/ErsatzTV.Infrastructure.Sqlite/Data/SqliteErrorClassifier.cs new file mode 100644 index 000000000..2ad1a9534 --- /dev/null +++ b/ErsatzTV.Infrastructure.Sqlite/Data/SqliteErrorClassifier.cs @@ -0,0 +1,22 @@ +using Microsoft.Data.Sqlite; +using Microsoft.EntityFrameworkCore; + +namespace ErsatzTV.Infrastructure.Sqlite.Data; + +/// +/// Classifies SQLite provider exceptions wrapped by EF Core. Wired to +/// at startup. +/// +public static class SqliteErrorClassifier +{ + // SQLITE_CONSTRAINT (primary result code 19), narrowed to the two extended codes that mean a + // duplicate key: SQLITE_CONSTRAINT_PRIMARYKEY (1555) and SQLITE_CONSTRAINT_UNIQUE (2067). Other + // constraint failures (FK 787, NOT NULL 1299, CHECK 275) are deliberately NOT treated as duplicates. + private const int SqliteConstraint = 19; + private const int SqliteConstraintPrimaryKey = 1555; + private const int SqliteConstraintUnique = 2067; + + public static bool IsUniqueConstraintViolation(DbUpdateException ex) => + ex.InnerException is SqliteException { SqliteErrorCode: SqliteConstraint } inner + && inner.SqliteExtendedErrorCode is SqliteConstraintPrimaryKey or SqliteConstraintUnique; +} diff --git a/ErsatzTV.Infrastructure/Data/TvContext.cs b/ErsatzTV.Infrastructure/Data/TvContext.cs index b226b4016..2e014ebc9 100644 --- a/ErsatzTV.Infrastructure/Data/TvContext.cs +++ b/ErsatzTV.Infrastructure/Data/TvContext.cs @@ -26,6 +26,16 @@ public class TvContext : DbContext public static string CaseInsensitiveCollation { get; set; } = "NOCASE"; public static bool IsSqlite { get; set; } + /// + /// Recognizes a provider-specific unique / primary-key constraint violation surfaced as a + /// (SQLite error 19 / MySQL 1062). Set at startup by the active + /// provider's wiring, mirroring / . Used to + /// turn a concurrent duplicate insert (a row we membership-checked absent, inserted first by a racing + /// request) into an idempotent no-op instead of a 500 — see ConcurrencyExtensions and #308. + /// Defaults to a conservative "no" so an unwired provider never silently swallows a save failure. + /// + public static Func IsUniqueConstraintViolation { get; set; } = static _ => false; + public IDbConnection Connection => Database.GetDbConnection(); public DbSet ConfigElements { get; set; } diff --git a/ErsatzTV.Tests/Application/MediaCollections/AddToCollectionIdempotencyConcurrencyTests.cs b/ErsatzTV.Tests/Application/MediaCollections/AddToCollectionIdempotencyConcurrencyTests.cs new file mode 100644 index 000000000..93eb48c87 --- /dev/null +++ b/ErsatzTV.Tests/Application/MediaCollections/AddToCollectionIdempotencyConcurrencyTests.cs @@ -0,0 +1,256 @@ +using System.Threading.Channels; +using ErsatzTV.Application; +using ErsatzTV.Application.MediaCollections; +using ErsatzTV.Application.Search; +using ErsatzTV.Core; +using ErsatzTV.Core.Domain; +using ErsatzTV.Core.Interfaces.Repositories; +using ErsatzTV.Infrastructure.Data; +using ErsatzTV.Infrastructure.Sqlite.Data; +using ErsatzTV.Tests.Support; +using LanguageExt; +using Microsoft.Data.Sqlite; +using Microsoft.EntityFrameworkCore; +using Microsoft.EntityFrameworkCore.Diagnostics; +using NSubstitute; +using NUnit.Framework; +using Shouldly; +using Unit = LanguageExt.Unit; + +namespace ErsatzTV.Tests.Application.MediaCollections; + +/// +/// #308: two concurrent adds of the same item both membership-check it absent, both insert the same +/// CollectionItem composite key, and the loser used to 500 on the unhandled +/// (SQLite 19 / MySQL 1062) that SaveChangesForcingVersion +/// does not catch. The fix (TrySaveChangesForcingVersion) makes the loser an idempotent no-op +/// for the single-item handlers, and for the bulk handler retries against fresh membership so the +/// non-colliding items are not dropped. +/// +[TestFixture] +public class AddToCollectionIdempotencyConcurrencyTests +{ + private InMemoryTvContext _db = null!; + + [SetUp] + public async Task SetUp() => _db = await InMemoryTvContext.CreateAsync(); + + [TearDown] + public async Task TearDown() => await _db.DisposeAsync(); + + private static async Task SeedCollectionWithMovies(Func createContext, int collectionId, params int[] movieIds) + { + await using TvContext context = createContext(); + + var library = new LocalLibrary + { + Id = 1, Name = "Library", MediaKind = LibraryMediaKind.Movies, Paths = [] + }; + var path = new LibraryPath { Id = 1, Path = "/media", Library = library, LibraryFolders = [], MediaItems = [] }; + library.Paths.Add(path); + + foreach (int movieId in movieIds) + { + context.Movies.Add(new Movie + { + Id = movieId, + LibraryPath = path, + Collections = [], + CollectionItems = [], + TraktListItems = [], + MovieMetadata = + [ + new MovieMetadata + { + Title = $"Movie {movieId}", SortTitle = $"Movie {movieId}", Artwork = [], Genres = [], + Tags = [], Studios = [], Actors = [], Guids = [], Subtitles = [], Directors = [], Writers = [] + } + ], + MediaVersions = [new MediaVersion { Duration = TimeSpan.FromMinutes(30) }] + }); + } + + context.Collections.Add(new Collection + { + Id = collectionId, Name = "Manual", Version = 1, MediaItems = [], CollectionItems = [] + }); + + await context.SaveChangesAsync(); + } + + private static async Task CollectionItemCount(Func createContext, int collectionId, int mediaItemId) + { + await using TvContext context = createContext(); + return await context.CollectionItems + .CountAsync(ci => ci.CollectionId == collectionId && ci.MediaItemId == mediaItemId); + } + + private static (IMediaCollectionRepository repo, Channel worker, + ChannelWriter search) Deps() + { + IMediaCollectionRepository repo = Substitute.For(); + repo.PlayoutIdsUsingCollection(Arg.Any()).Returns([1]); + Channel worker = + System.Threading.Channels.Channel.CreateUnbounded(); + ChannelWriter search = + System.Threading.Channels.Channel.CreateUnbounded().Writer; + return (repo, worker, search); + } + + // Simulates the concurrent "winner": exactly once, on a SEPARATE connection, insert the same composite + // key and commit — AFTER the intercepted context has read (stale) membership but BEFORE its own INSERT + // runs. This is what interposes the race deterministically. + private sealed class InsertConflictingItemOnce(SharedCacheTvContext db, int collectionId, int mediaItemId) + : SaveChangesInterceptor + { + private int _fired; + + public override async ValueTask> SavingChangesAsync( + DbContextEventData eventData, + InterceptionResult result, + CancellationToken cancellationToken = default) + { + if (Interlocked.Exchange(ref _fired, 1) == 0) + { + await using SqliteConnection connection = db.OpenConnection(); + await using SqliteCommand command = connection.CreateCommand(); + command.CommandText = + "INSERT INTO \"CollectionItem\" (\"CollectionId\", \"MediaItemId\", \"CustomIndex\") " + + $"VALUES ({collectionId}, {mediaItemId}, NULL)"; + await command.ExecuteNonQueryAsync(cancellationToken); + } + + return result; + } + } + + // ----- Non-vacuous anchor + extension-level unit tests (single shared connection) ----- + + // Proves the race really does throw a DbUpdateException the classifier recognizes — i.e. the pre-fix + // 500 was real and the fix's catch filter is not a no-op. This is the negative control anchoring the + // extension tests below. + [Test] + public async Task Duplicate_CollectionItem_Insert_Throws_A_Classified_UniqueViolation() + { + await SeedCollectionWithMovies(_db.CreateContext, 1, 10); + + await using TvContext stale = _db.CreateContext(); + Collection tracked = await stale.Collections.Include(c => c.MediaItems).SingleAsync(c => c.Id == 1); + Movie movie = await stale.Movies.SingleAsync(m => m.Id == 10); + + // A racer inserts the (1, 10) CollectionItem and commits, then the stale context tries the same key. + await InsertRow(_db.CreateContext, 1, 10); + tracked.MediaItems.Add(movie); + + DbUpdateException ex = await Should.ThrowAsync(() => stale.SaveChangesAsync()); + SqliteErrorClassifier.IsUniqueConstraintViolation(ex).ShouldBeTrue(); + + // And the classifier does not blanket-return true for an unrelated failure. + SqliteErrorClassifier.IsUniqueConstraintViolation( + new DbUpdateException("nope", new InvalidOperationException())).ShouldBeFalse(); + } + + [Test] + public async Task TrySaveChangesForcingVersion_Swallows_The_Duplicate_And_Returns_False() + { + await SeedCollectionWithMovies(_db.CreateContext, 1, 10); + + await using TvContext stale = _db.CreateContext(); + Collection tracked = await stale.Collections.Include(c => c.MediaItems).SingleAsync(c => c.Id == 1); + Movie movie = await stale.Movies.SingleAsync(m => m.Id == 10); + + await InsertRow(_db.CreateContext, 1, 10); + tracked.MediaItems.Add(movie); + tracked.Version++; + + bool committed = await stale.TrySaveChangesForcingVersion(CancellationToken.None); + + committed.ShouldBeFalse(); // the loser's save was swallowed, not thrown + (await CollectionItemCount(_db.CreateContext, 1, 10)).ShouldBe(1); // exactly one row, no corruption + } + + [Test] + public async Task TrySaveChangesForcingVersion_Rethrows_When_The_Provider_Does_Not_Classify_It() + { + // Guard against an over-broad catch: with the classifier saying "not a unique violation", the + // duplicate must still propagate rather than be silently swallowed. + Func original = TvContext.IsUniqueConstraintViolation; + try + { + await SeedCollectionWithMovies(_db.CreateContext, 1, 10); + + await using TvContext stale = _db.CreateContext(); + Collection tracked = await stale.Collections.Include(c => c.MediaItems).SingleAsync(c => c.Id == 1); + Movie movie = await stale.Movies.SingleAsync(m => m.Id == 10); + + await InsertRow(_db.CreateContext, 1, 10); + tracked.MediaItems.Add(movie); + tracked.Version++; + + TvContext.IsUniqueConstraintViolation = _ => false; + await Should.ThrowAsync(() => stale.TrySaveChangesForcingVersion(CancellationToken.None)); + } + finally + { + TvContext.IsUniqueConstraintViolation = original; + } + } + + // ----- End-to-end handler tests over a real cross-connection race (shared cache) ----- + + [Test] + public async Task SingleItem_Handler_Losing_The_Cross_Connection_Race_Is_A_NoOp_Not_A_500() + { + await using SharedCacheTvContext db = await SharedCacheTvContext.CreateAsync("etv308-single"); + await SeedCollectionWithMovies(db.CreateContext, 1, 10); + + (IMediaCollectionRepository repo, Channel worker, + ChannelWriter search) = Deps(); + + // The interceptor inserts (1, 10) on another connection during this handler's save. + var racer = new InsertConflictingItemOnce(db, 1, 10); + var handler = new AddMovieToCollectionHandler(db.Factory(racer), repo, worker.Writer, search); + + Either result = await handler.Handle(new AddMovieToCollection(1, 10), CancellationToken.None); + + result.IsRight.ShouldBeTrue(); // no 500 + (await CollectionItemCount(db.CreateContext, 1, 10)).ShouldBe(1); // one row, no duplicate + worker.Reader.Count.ShouldBe(0); // the loser does not fan out a rebuild (the winner already did) + } + + [Test] + public async Task Bulk_Handler_Losing_The_Cross_Connection_Race_Retries_And_Keeps_NonColliding_Items() + { + await using SharedCacheTvContext db = await SharedCacheTvContext.CreateAsync("etv308-bulk"); + await SeedCollectionWithMovies(db.CreateContext, 1, 10, 11); + + (IMediaCollectionRepository repo, Channel worker, + ChannelWriter search) = Deps(); + + var movieRepo = Substitute.For(); + movieRepo.AllMoviesExist(Arg.Any>()).Returns(true); + var tvRepo = Substitute.For(); + tvRepo.AllShowsExist(Arg.Any>()).Returns(true); + tvRepo.AllSeasonsExist(Arg.Any>()).Returns(true); + tvRepo.AllEpisodesExist(Arg.Any>()).Returns(true); + + // Item 10 is inserted by the racer during the first attempt's save, colliding and rolling back the + // whole [10, 11] batch; the retry must recompute membership and still add 11. + var racer = new InsertConflictingItemOnce(db, 1, 10); + var handler = new AddItemsToCollectionHandler(db.Factory(racer), repo, movieRepo, tvRepo, worker.Writer, search); + + var request = new AddItemsToCollection(1, [10, 11], [], [], [], [], [], [], [], [], []); + Either result = await handler.Handle(request, CancellationToken.None); + + result.IsRight.ShouldBeTrue(); + (await CollectionItemCount(db.CreateContext, 1, 10)).ShouldBe(1); // the racer's row — not dropped/duped + (await CollectionItemCount(db.CreateContext, 1, 11)).ShouldBe(1); // NOT dropped by the rolled-back attempt + } + + private static async Task InsertRow(Func createContext, int collectionId, int mediaItemId) + { + await using TvContext context = createContext(); + context.Add(new CollectionItem { CollectionId = collectionId, MediaItemId = mediaItemId, CustomIndex = null }); + await context.SaveChangesAsync(); + } +} diff --git a/ErsatzTV.Tests/Support/InMemoryTvContext.cs b/ErsatzTV.Tests/Support/InMemoryTvContext.cs index 93def6448..5dcb34160 100644 --- a/ErsatzTV.Tests/Support/InMemoryTvContext.cs +++ b/ErsatzTV.Tests/Support/InMemoryTvContext.cs @@ -1,5 +1,6 @@ using ErsatzTV.Infrastructure; using ErsatzTV.Infrastructure.Data; +using ErsatzTV.Infrastructure.Sqlite.Data; using Microsoft.Data.Sqlite; using Microsoft.EntityFrameworkCore; using Microsoft.Extensions.Logging.Abstractions; @@ -29,6 +30,7 @@ public sealed class InMemoryTvContext : IAsyncDisposable public static async Task CreateAsync() { TvContext.IsSqlite = true; + TvContext.IsUniqueConstraintViolation = SqliteErrorClassifier.IsUniqueConstraintViolation; var connection = new SqliteConnection("Data Source=:memory:;Foreign Keys=False"); await connection.OpenAsync(); diff --git a/ErsatzTV.Tests/Support/SharedCacheTvContext.cs b/ErsatzTV.Tests/Support/SharedCacheTvContext.cs new file mode 100644 index 000000000..970a339bd --- /dev/null +++ b/ErsatzTV.Tests/Support/SharedCacheTvContext.cs @@ -0,0 +1,89 @@ +using ErsatzTV.Infrastructure; +using ErsatzTV.Infrastructure.Data; +using ErsatzTV.Infrastructure.Sqlite.Data; +using Microsoft.Data.Sqlite; +using Microsoft.EntityFrameworkCore; +using Microsoft.EntityFrameworkCore.Diagnostics; +using Microsoft.Extensions.Logging.Abstractions; + +namespace ErsatzTV.Tests.Support; + +/// +/// A multi-connection in-memory SQLite harness (shared cache) so several +/// instances on DISTINCT connections hit the same database and real unique-constraint enforcement +/// applies across connections. Unlike (a single shared connection), +/// this can reproduce a genuine cross-connection write race — needed to exercise the #308 composite-PK +/// collision through the real handler. A keep-alive connection holds the shared in-memory DB open for +/// the harness lifetime. +/// +public sealed class SharedCacheTvContext : IAsyncDisposable +{ + private readonly string _connectionString; + private readonly SqliteConnection _keepAlive; + + private SharedCacheTvContext(string connectionString, SqliteConnection keepAlive) + { + _connectionString = connectionString; + _keepAlive = keepAlive; + } + + public static async Task CreateAsync(string name) + { + TvContext.IsSqlite = true; + TvContext.IsUniqueConstraintViolation = SqliteErrorClassifier.IsUniqueConstraintViolation; + + // Foreign keys OFF (connection-string keyword, inherited by every connection) so partial object + // graphs can be seeded without satisfying every FK — matches InMemoryTvContext. The composite-PK / + // UNIQUE constraint under test is enforced regardless of this pragma. + var connectionString = $"Data Source={name};Mode=Memory;Cache=Shared;Foreign Keys=False"; + var keepAlive = new SqliteConnection(connectionString); + await keepAlive.OpenAsync(); + + await using (TvContext context = Create(BuildOptions(connectionString, null))) + { + await context.Database.EnsureCreatedAsync(); + } + + return new SharedCacheTvContext(connectionString, keepAlive); + } + + /// A factory whose contexts each open their own connection to the shared DB. + public IDbContextFactory Factory(IInterceptor? interceptor = null) => + new PerConnectionFactory(_connectionString, interceptor); + + public TvContext CreateContext() => Create(BuildOptions(_connectionString, null)); + + /// A freshly opened raw connection to the shared DB (used to stage a concurrent writer). + public SqliteConnection OpenConnection() + { + var connection = new SqliteConnection(_connectionString); + connection.Open(); + return connection; + } + + public async ValueTask DisposeAsync() => await _keepAlive.DisposeAsync(); + + private static DbContextOptions BuildOptions(string connectionString, IInterceptor? interceptor) + { + DbContextOptionsBuilder builder = new DbContextOptionsBuilder() + .UseSqlite(connectionString); + if (interceptor is not null) + { + builder.AddInterceptors(interceptor); + } + + return builder.Options; + } + + private static TvContext Create(DbContextOptions options) => + new( + options, + NullLoggerFactory.Instance, + new SlowQueryInterceptor(NullLogger.Instance)); + + private sealed class PerConnectionFactory(string connectionString, IInterceptor? interceptor) + : IDbContextFactory + { + public TvContext CreateDbContext() => Create(BuildOptions(connectionString, interceptor)); + } +} diff --git a/ErsatzTV/Startup.cs b/ErsatzTV/Startup.cs index dd5aebf21..c147544ca 100644 --- a/ErsatzTV/Startup.cs +++ b/ErsatzTV/Startup.cs @@ -71,6 +71,7 @@ using ErsatzTV.Infrastructure.Locking; using ErsatzTV.Infrastructure.Metadata; using ErsatzTV.Infrastructure.Plex; using ErsatzTV.Infrastructure.Runtime; +using ErsatzTV.Infrastructure.MySql.Data; using ErsatzTV.Infrastructure.Scheduling; using ErsatzTV.Infrastructure.Scripting; using ErsatzTV.Infrastructure.Search; @@ -647,6 +648,7 @@ public class Startup TvContext.LastInsertedRowId = "last_insert_rowid()"; TvContext.CaseInsensitiveCollation = "NOCASE"; + TvContext.IsUniqueConstraintViolation = SqliteErrorClassifier.IsUniqueConstraintViolation; SqlMapper.AddTypeHandler(new DateTimeOffsetHandler()); SqlMapper.AddTypeHandler(new GuidHandler()); @@ -657,6 +659,7 @@ public class Startup { TvContext.LastInsertedRowId = "last_insert_id()"; TvContext.CaseInsensitiveCollation = "utf8mb4_general_ci"; + TvContext.IsUniqueConstraintViolation = MySqlErrorClassifier.IsUniqueConstraintViolation; } Log.Logger.Information("Transcode folder is {Folder}", FileSystemLayout.TranscodeFolder); diff --git a/docs/api-conventions.md b/docs/api-conventions.md index 864d7e9d7..45855c951 100644 --- a/docs/api-conventions.md +++ b/docs/api-conventions.md @@ -577,11 +577,25 @@ rotate the editor ETag). **No-op idempotence (the trap):** these handlers gate t fan-out on `SaveChanges() > 0`; an *unconditional* bump makes that gate always-true, so an idempotent re-add / same-value re-submit would fire spurious rebuilds. Each therefore short-circuits a genuine no-op **before** the bump — the Add handlers by an explicit membership check (also fixing the latent duplicate-`CollectionItem` -insert on a *sequential* re-add; two *concurrent* adds of the same item can still both pass the check and the -loser 500s on the composite-PK violation — a narrow, pre-existing race, tracked as #308), the scalar -writers by `ChangeTracker.HasChanges()` — so a no-op neither bumps nor rebuilds. This is an -invalidation-completeness refinement; the primary endpoints' own bump+guard already covered the two-tab -lost-update the contract targets. +insert on a *sequential* re-add), the scalar writers by `ChangeTracker.HasChanges()` — so a no-op neither +bumps nor rebuilds. This is an invalidation-completeness refinement; the primary endpoints' own bump+guard +already covered the two-tab lost-update the contract targets. + +**Idempotent insert under concurrency (#308).** The membership pre-check is not atomic with the insert, so +two *concurrent* adds of the same item both observe it absent, both stage the `CollectionItem` composite key, +and the loser's `SaveChangesForcingVersion` throws a unique/PK-violation `DbUpdateException` (SQLite error 19 / +MySQL 1062) it does not catch → a 500. The `Add*ToCollection` family therefore saves through +**`ConcurrencyExtensions.TrySaveChangesForcingVersion`** (a `bool`-returning sibling of `SaveChangesForcingVersion`) +which catches *only* that classified violation and returns `false`. The single-item handlers treat `false` as an +idempotent **no-op** (the racing winner already inserted the row, rotated the ETag, and fanned out the rebuild); +the bulk `AddItemsToCollection` handler instead **retries** on a fresh context against recomputed membership so +the non-colliding items in the batch are not dropped (bounded loop; the common no-collision path runs once). +The provider-specific classifier is wired the same way as the other provider statics on `TvContext` — a settable +`TvContext.IsUniqueConstraintViolation` delegate pointed at `SqliteErrorClassifier` / `MySqlErrorClassifier` +(`ErsatzTV.Infrastructure.Sqlite/MySql.Data`) from `Startup.cs`, defaulting to a conservative "no" so an unwired +provider never silently swallows a save failure. `Add*ToPlaylist` is **not** affected: `PlaylistItem` has its own +identity PK and no unique index on `(PlaylistId, MediaItemId)` — a playlist may legitimately contain the same item +more than once, so there is no constraint to violate. ## 7b. Post-commit side effects run on `CancellationToken.None` diff --git a/docs/decisions/optimistic-concurrency.md b/docs/decisions/optimistic-concurrency.md index 3a8484e8c..4a54a1532 100644 --- a/docs/decisions/optimistic-concurrency.md +++ b/docs/decisions/optimistic-concurrency.md @@ -17,6 +17,7 @@ cross-editor ETag rotation). Refs #197. - [2026-07-12 (#269 — non-If-Match root writers force-write past a concurrent Version bump)](#2026-07-12-269--non-if-match-root-writers-force-write-past-a-concurrent-version-bump) - [2026-07-12 — Cross-editor ETag rotation completed for Collection/Playout config siblings (#269)](#2026-07-12--cross-editor-etag-rotation-completed-for-collectionplayout-config-siblings-269) - [2026-07-12 — If-Match evaluates per RFC 7232: valid-but-non-matching → 412, only grammar violations → 400 (#265)](#2026-07-12--if-match-evaluates-per-rfc-7232-valid-but-non-matching--412-only-grammar-violations--400-265) +- [2026-07-18 — Concurrent same-item add is idempotent, not a 500: catch the unique-violation per provider (#308)](#2026-07-18--concurrent-same-item-add-is-idempotent-not-a-500-catch-the-unique-violation-per-provider-308) --- @@ -254,3 +255,36 @@ worried about was already correct (each handler loads/validates → 404 before ` first-party SPA only ever echoes the single canonical strong tag we emit, so no shipped client changes behavior; the change only makes a hand-written/tooling `If-Match` get the RFC-correct status. Docs: `api-conventions.md` §7a. Refs #265 #253 #197. + +## 2026-07-18 — Concurrent same-item add is idempotent, not a 500: catch the unique-violation per provider (#308) + +**Decision.** The `Add*ToCollection` family's membership pre-check (#269) is not atomic with the insert, so two +*concurrent* adds of the same item both observe it absent and both stage the `CollectionItem` composite key; the +loser's `SaveChangesForcingVersion` threw a unique/PK-violation `DbUpdateException` (SQLite error 19 / MySQL 1062) +it did not catch → **500**. We now treat that loss as an **idempotent no-op**, not an error: the desired end state +(the item is a member) already holds because the racing winner inserted it, rotated the ETag, and fanned out the +rebuild. + +**Mechanism.** A `bool`-returning sibling `ConcurrencyExtensions.TrySaveChangesForcingVersion` wraps +`SaveChangesForcingVersion` and catches *only* a classified unique/PK violation, returning `false`. The 10 +single-item handlers return `Unit.Default` on `false` (skip the reindex/rebuild fan-out — the winner did it). The +bulk `AddItemsToCollection` handler cannot no-op — that would silently drop the non-colliding items when a batch +partially overlaps a concurrent add — so it **retries** on a fresh context against recomputed membership (bounded +loop; the common no-collision path runs once). + +**Provider seam.** Detection is provider-specific but the Application layer must not reference the provider +packages, so it follows the existing `TvContext` static-provider-config idiom (`IsSqlite`, `LastInsertedRowId`): a +settable `TvContext.IsUniqueConstraintViolation` delegate, pointed at `SqliteErrorClassifier` (extended codes 1555 +PK / 2067 UNIQUE) or `MySqlErrorClassifier` (`Number == 1062`) from `Startup.cs`, defaulting to a conservative +"no" so an unwired provider never silently swallows a save failure. Chosen over DI to avoid threading a new +service through 11 handlers, and because the provider discriminator already lives as a `TvContext` static. + +**Scope boundary.** `Add*ToPlaylist` is deliberately **untouched**: `PlaylistItem` has its own identity PK and no +unique index on `(PlaylistId, MediaItemId)` — a playlist may legitimately contain the same item more than once, so +there is no constraint to violate. + +**Tests.** A negative-control anchor proves the race genuinely throws a classified `DbUpdateException`; the fix's +end-to-end handler tests reproduce a *real cross-connection* race via a shared-cache SQLite harness + a +`SavingChanges` interceptor that inserts the conflicting row on another connection mid-save (the single-connection +in-memory fixture cannot). Every fix-dependent test was verified to fail with the catch disabled. Mechanics: +`api-conventions.md` §7a ("Idempotent insert under concurrency"). Refs #308 #269 #253.