fix(308): idempotent concurrent Add*ToCollection instead of a composite-PK 500
Build ErsatzTV Image / CI image pin matches docker/ci (pull_request) Successful in 8s
Build ErsatzTV Image / API docs in sync (OpenAPI + endpoint index) (pull_request) Successful in 12s
Build ErsatzTV Image / Docs update reminder (pull_request) Successful in 8s
Build ErsatzTV Image / decisions.md append-only (pull_request) Successful in 40s
Build ErsatzTV Image / Formatting (changed .cs conform to .editorconfig) (pull_request) Successful in 7m25s
Build ErsatzTV Image / Functional E2E (curl contracts) (pull_request) Successful in 14m59s
Build ErsatzTV Image / EF migration integrity (SQLite + MySql) (pull_request) Successful in 19m3s
Build ErsatzTV Image / Build & test (.NET) (pull_request) Successful in 19m47s
Build ErsatzTV Image / Build & push image (amd64) (pull_request) Has been skipped

Two concurrent adds of the same item both membership-check it absent, both
insert the CollectionItem composite key, and the loser's
SaveChangesForcingVersion threw an uncaught DbUpdateException (SQLite 19 /
MySQL 1062) -> 500. Now the loser is an idempotent no-op.

- ConcurrencyExtensions.TrySaveChangesForcingVersion: bool-returning sibling
  that catches only a classified unique/PK violation and returns false.
- 10 single-item Add*ToCollection handlers: return Unit.Default (no-op, skip
  fan-out) on false — the racing winner already inserted + rotated + rebuilt.
- Bulk AddItemsToCollection: retry on a fresh context against recomputed
  membership so a partial-overlap collision doesn't drop the non-colliding
  items (bounded loop; common no-collision path runs once).
- Provider detection via a TvContext.IsUniqueConstraintViolation static
  delegate (matches the existing IsSqlite/LastInsertedRowId provider seam),
  wired from Startup to SqliteErrorClassifier / MySqlErrorClassifier.
- Add*ToPlaylist is NOT affected (PlaylistItem has its own identity PK; a
  playlist may legitimately contain the same item more than once).

Tests: a negative-control anchor proves the race genuinely throws a classified
exception; end-to-end handler tests reproduce a real cross-connection race via
a shared-cache SQLite harness + a SavingChanges interceptor (the single-conn
in-memory fixture cannot). Every fix-dependent test verified to fail with the
catch disabled.

Docs: api-conventions.md §7a (idempotent insert under concurrency) +
decisions/optimistic-concurrency.md.

fixes #308

Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>
This commit is contained in:
2026-07-18 13:02:14 +02:00
co-authored by Claude Opus 4.8
parent 3a463db36a
commit 2281f2e764
21 changed files with 611 additions and 40 deletions
@@ -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
}
}
/// <summary>
/// Like <see cref="SaveChangesForcingVersion" />, but additionally treats a unique / primary-key
/// constraint violation as an idempotent no-op: returns <c>false</c> instead of throwing when the
/// save fails because a concurrent request inserted a row we had membership-checked absent (the
/// composite-PK race on <c>CollectionItem</c> — issue #308). A <c>false</c> means "the desired row
/// already exists because a racing writer won; the winner ran the ETag rotation + fan-out, so skip
/// ours." <c>true</c> means our own change committed. Every other <see cref="DbUpdateException" />
/// (and the genuine deleted-row concurrency conflict rethrown by <see cref="SaveChangesForcingVersion" />)
/// still propagates. The only insert these callers stage is the <c>CollectionItem</c> join row, so the
/// sole unique/PK constraint that can fire here is that composite key.
/// </summary>
public static async Task<bool> TrySaveChangesForcingVersion(
this DbContext dbContext,
CancellationToken cancellationToken)
{
try
{
await dbContext.SaveChangesForcingVersion(cancellationToken);
return true;
}
catch (DbUpdateException ex) when (TvContext.IsUniqueConstraintViolation(ex))
{
return false;
}
}
/// <summary>
/// Persist pending changes, mapping the EF optimistic-concurrency failure to
/// <see cref="PreconditionFailedError" /> (→ 412). When a versioned root carries an
@@ -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);
@@ -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);
@@ -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<AddImageToCollection,
// 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.Image.Id]), CancellationToken.None);
@@ -1,4 +1,4 @@
using System.Threading.Channels;
using System.Threading.Channels;
using ErsatzTV.Application.Playouts;
using ErsatzTV.Application.Search;
using ErsatzTV.Core;
@@ -38,23 +38,52 @@ public class AddItemsToCollectionHandler :
_searchChannel = searchChannel;
}
// A duplicate-key race can roll back the whole batch (#308); recompute membership from a fresh
// context and retry with only the still-missing items. Bounded to avoid a livelock — the common
// no-collision path runs the loop body exactly once.
private const int MaxDuplicateRetries = 5;
public async Task<Either<BaseError, Unit>> Handle(
AddItemsToCollection request,
CancellationToken cancellationToken)
{
await using TvContext dbContext = await _dbContextFactory.CreateDbContextAsync(cancellationToken);
Option<Collection> 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<Collection> 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<BaseError, bool> attemptResult = await maybeCollection.Match(
Some: async collection =>
{
Validation<BaseError, Collection> validation = await Validate(dbContext, request, collection, cancellationToken);
return await validation.Apply(c => ApplyAddItemsRequest(dbContext, c, request, cancellationToken));
},
None: () => Task.FromResult<Either<BaseError, bool>>(
new NotFoundError($"Collection {request.CollectionId} does not exist.")));
if (attemptResult.IsLeft)
{
Validation<BaseError, Collection> validation = await Validate(dbContext, request, collection, cancellationToken);
return await validation.Apply(c => ApplyAddItemsRequest(dbContext, c, request, cancellationToken));
},
None: () => Task.FromResult<Either<BaseError, Unit>>(
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<Unit> ApplyAddItemsRequest(
private async Task<bool> 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<MediaItem> 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<Validation<BaseError, Collection>> Validate(
@@ -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);
@@ -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);
@@ -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);
@@ -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);
@@ -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);
@@ -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);
@@ -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);
@@ -0,0 +1,17 @@
using Microsoft.EntityFrameworkCore;
using MySqlConnector;
namespace ErsatzTV.Infrastructure.MySql.Data;
/// <summary>
/// Classifies MySQL provider exceptions wrapped by EF Core. Wired to
/// <see cref="ErsatzTV.Infrastructure.Data.TvContext.IsUniqueConstraintViolation" /> at startup.
/// </summary>
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 };
}
@@ -0,0 +1,22 @@
using Microsoft.Data.Sqlite;
using Microsoft.EntityFrameworkCore;
namespace ErsatzTV.Infrastructure.Sqlite.Data;
/// <summary>
/// Classifies SQLite provider exceptions wrapped by EF Core. Wired to
/// <see cref="ErsatzTV.Infrastructure.Data.TvContext.IsUniqueConstraintViolation" /> at startup.
/// </summary>
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;
}
+10
View File
@@ -26,6 +26,16 @@ public class TvContext : DbContext
public static string CaseInsensitiveCollation { get; set; } = "NOCASE";
public static bool IsSqlite { get; set; }
/// <summary>
/// Recognizes a provider-specific unique / primary-key constraint violation surfaced as a
/// <see cref="DbUpdateException" /> (SQLite error 19 / MySQL 1062). Set at startup by the active
/// provider's wiring, mirroring <see cref="IsSqlite" /> / <see cref="LastInsertedRowId" />. 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 <c>ConcurrencyExtensions</c> and #308.
/// Defaults to a conservative "no" so an unwired provider never silently swallows a save failure.
/// </summary>
public static Func<DbUpdateException, bool> IsUniqueConstraintViolation { get; set; } = static _ => false;
public IDbConnection Connection => Database.GetDbConnection();
public DbSet<ConfigElement> ConfigElements { get; set; }
@@ -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;
/// <summary>
/// #308: two concurrent adds of the same item both membership-check it absent, both insert the same
/// <c>CollectionItem</c> composite key, and the loser used to 500 on the unhandled
/// <see cref="DbUpdateException" /> (SQLite 19 / MySQL 1062) that <c>SaveChangesForcingVersion</c>
/// does not catch. The fix (<c>TrySaveChangesForcingVersion</c>) 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.
/// </summary>
[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<TvContext> 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<int> CollectionItemCount(Func<TvContext> 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<IBackgroundServiceRequest> worker,
ChannelWriter<ISearchIndexBackgroundServiceRequest> search) Deps()
{
IMediaCollectionRepository repo = Substitute.For<IMediaCollectionRepository>();
repo.PlayoutIdsUsingCollection(Arg.Any<int>()).Returns([1]);
Channel<IBackgroundServiceRequest> worker =
System.Threading.Channels.Channel.CreateUnbounded<IBackgroundServiceRequest>();
ChannelWriter<ISearchIndexBackgroundServiceRequest> search =
System.Threading.Channels.Channel.CreateUnbounded<ISearchIndexBackgroundServiceRequest>().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<InterceptionResult<int>> SavingChangesAsync(
DbContextEventData eventData,
InterceptionResult<int> 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<DbUpdateException>(() => 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<DbUpdateException, bool> 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<DbUpdateException>(() => 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<IBackgroundServiceRequest> worker,
ChannelWriter<ISearchIndexBackgroundServiceRequest> 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<BaseError, Unit> 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<IBackgroundServiceRequest> worker,
ChannelWriter<ISearchIndexBackgroundServiceRequest> search) = Deps();
var movieRepo = Substitute.For<IMovieRepository>();
movieRepo.AllMoviesExist(Arg.Any<List<int>>()).Returns(true);
var tvRepo = Substitute.For<ITelevisionRepository>();
tvRepo.AllShowsExist(Arg.Any<List<int>>()).Returns(true);
tvRepo.AllSeasonsExist(Arg.Any<List<int>>()).Returns(true);
tvRepo.AllEpisodesExist(Arg.Any<List<int>>()).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<BaseError, Unit> 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<TvContext> createContext, int collectionId, int mediaItemId)
{
await using TvContext context = createContext();
context.Add(new CollectionItem { CollectionId = collectionId, MediaItemId = mediaItemId, CustomIndex = null });
await context.SaveChangesAsync();
}
}
@@ -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<InMemoryTvContext> CreateAsync()
{
TvContext.IsSqlite = true;
TvContext.IsUniqueConstraintViolation = SqliteErrorClassifier.IsUniqueConstraintViolation;
var connection = new SqliteConnection("Data Source=:memory:;Foreign Keys=False");
await connection.OpenAsync();
@@ -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;
/// <summary>
/// A multi-connection in-memory SQLite harness (shared cache) so several <see cref="TvContext" />
/// instances on DISTINCT connections hit the same database and real unique-constraint enforcement
/// applies across connections. Unlike <see cref="InMemoryTvContext" /> (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.
/// </summary>
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<SharedCacheTvContext> 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);
}
/// <summary>A factory whose contexts each open their own connection to the shared DB.</summary>
public IDbContextFactory<TvContext> Factory(IInterceptor? interceptor = null) =>
new PerConnectionFactory(_connectionString, interceptor);
public TvContext CreateContext() => Create(BuildOptions(_connectionString, null));
/// <summary>A freshly opened raw connection to the shared DB (used to stage a concurrent writer).</summary>
public SqliteConnection OpenConnection()
{
var connection = new SqliteConnection(_connectionString);
connection.Open();
return connection;
}
public async ValueTask DisposeAsync() => await _keepAlive.DisposeAsync();
private static DbContextOptions<TvContext> BuildOptions(string connectionString, IInterceptor? interceptor)
{
DbContextOptionsBuilder<TvContext> builder = new DbContextOptionsBuilder<TvContext>()
.UseSqlite(connectionString);
if (interceptor is not null)
{
builder.AddInterceptors(interceptor);
}
return builder.Options;
}
private static TvContext Create(DbContextOptions<TvContext> options) =>
new(
options,
NullLoggerFactory.Instance,
new SlowQueryInterceptor(NullLogger<SlowQueryInterceptor>.Instance));
private sealed class PerConnectionFactory(string connectionString, IInterceptor? interceptor)
: IDbContextFactory<TvContext>
{
public TvContext CreateDbContext() => Create(BuildOptions(connectionString, interceptor));
}
}
+3
View File
@@ -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);
+19 -5
View File
@@ -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`
+34
View File
@@ -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.