From f1fdc54f10e77266c348794c1b152b36278f8ce4 Mon Sep 17 00:00:00 2001 From: bitbiter-dev Date: Mon, 19 Oct 2026 13:57:46 +1300 Subject: [PATCH 1/3] feat(worker): Report duplicate content, never merge it (#36) Within a config, a copy of already-ingested content is now logged at Information with its content hash, storage config and both paths. Across one user's configs, SharedContentReporter runs after each crawl cycle over the configs just crawled and logs every pair of active rows that share a content hash. Duplicates across different users are not reported. A failed report is logged and never stops the crawl loop. Nothing is written to the database. Two new identical copies racing in the same crawl are first reported on the next crawl, the same delay the issue accepts for moves. Co-Authored-By: Claude Opus 5.5 --- CLAUDE.md | 3 +- .../Repository/MediaAssetRepositoryTests.cs | 92 +++++++++++++++++++ .../Data/Repository/MediaAssetRepository.cs | 24 +++++ .../Crawling/SharedContentReporterTests.cs | 82 +++++++++++++++++ .../Crawling/WorkerTests.cs | 64 ++++++++++++- .../WorkerServiceCollectionExtensionsTests.cs | 1 + .../ClassificationMiddlewareTests.cs | 23 ++++- .../Crawling/SharedContentReporter.cs | 40 ++++++++ src/Anichron.Worker/Crawling/Worker.cs | 18 ++++ .../WorkerServiceCollectionExtensions.cs | 1 + .../Classification/KnownFileReconciler.cs | 9 +- 11 files changed, 351 insertions(+), 6 deletions(-) create mode 100644 src/Anichron.Worker.Tests.Unit/Crawling/SharedContentReporterTests.cs create mode 100644 src/Anichron.Worker/Crawling/SharedContentReporter.cs diff --git a/CLAUDE.md b/CLAUDE.md index f37c17b4..938b6a2b 100644 --- a/CLAUDE.md +++ b/CLAUDE.md @@ -156,7 +156,8 @@ The API reads files only through two narrow stores in `Anichron.Infrastructure/S - Each `IImageProxyGenerator`, and the `BlurhashEncoder`, declares a `ProxyBound`; `ImageProxyMiddleware` decodes once to `Rgb24`, shrinks the shared image to each bound largest-first, and hands every step **its own copy** at that size. The encoder's string goes on the `IngestionContext` and the content writer (`ProcessedContentWriter.DescribeContent`) writes it to `Metadata`. Generators may crop/recolour/resize their copy freely; only the middleware resizes the shared image - FFmpeg transcodes video with runtime GPU detection: QuickSync (`h264_qsv`) → NVENC (`h264_nvenc`) → AMF (`h264_amf`) → software (`libx264`) - Burst detection (**planned, not yet implemented**): will group rapid-fire sequences and assign a `PrimaryAssetId` cover. The `Burst` entity and schema exist; no detection code is wired into the Worker yet. -- Reconciliation (**planned, Epic 4**): a periodic scan soft-deletes assets whose file is gone (interactions survive) and restores them when it reappears (#33, #34, #37). Moves are already handled at ingest (#35): when the hash matches an active row, `KnownFileReconciler` refreshes `LastSeenOnNas` if the path is unchanged (`Unchanged`), moves the row to the new path if its stored file is gone (`Moved`), and leaves it alone if the stored file still exists (`DuplicateCopy`, reported by #36). A Live Photo's MOV row gets the same treatment by its own hash. A move onto a path another row already holds, soft-deleted rows included (`PathTaken`), is logged as a warning and skipped; if a concurrent consumer takes the path after classification, the save fails and the item is retried on the next crawl. A file **edited in place** (hash unknown, but an active row holds the path: `ContentChanged`) is processed like new content and `ContentChangedWriter` updates that row: new hash, capture date, media type, `Metadata` and proxy rows (updated by type), while `Id`, interactions, `BurstId` and `PairedAssetId` are kept. A soft-deleted row at the path still blocks it (restore, #34). An edited Live Photo **MOV** keeps its stored row unchanged for now +- Reconciliation (**planned, Epic 4**): a periodic scan soft-deletes assets whose file is gone (interactions survive) and restores them when it reappears (#33, #34, #37). Moves are already handled at ingest (#35): when the hash matches an active row, `KnownFileReconciler` refreshes `LastSeenOnNas` if the path is unchanged (`Unchanged`), moves the row to the new path if its stored file is gone (`Moved`), and leaves it alone if the stored file still exists (`DuplicateCopy`, logged at `Information` with hash, config and both paths — #36). A Live Photo's MOV row gets the same treatment by its own hash. A move onto a path another row already holds, soft-deleted rows included (`PathTaken`), is logged as a warning and skipped; if a concurrent consumer takes the path after classification, the save fails and the item is retried on the next crawl. A file **edited in place** (hash unknown, but an active row holds the path: `ContentChanged`) is processed like new content and `ContentChangedWriter` updates that row: new hash, capture date, media type, `Metadata` and proxy rows (updated by type), while `Id`, interactions, `BurstId` and `PairedAssetId` are kept. A soft-deleted row at the path still blocks it (restore, #34). An edited Live Photo **MOV** keeps its stored row unchanged for now +- Duplicate content is **reported, never merged** (#36): besides `DuplicateCopy` within a config, `SharedContentReporter` runs after each crawl cycle and logs, once per pair, active rows of one user that share a `ContentHash` across the configs just crawled. Duplicates across different users are expected and never reported. Two new identical copies racing in one crawl are first reported on the next crawl - Failure ownership: `ProxyStagingWriter` is the single place a proxy reaches its final name (architecture rule `proxies-reach-final-name-only-in-staging-writer`) — it writes under a per-attempt `..tmp` suffix (so concurrent attempts on the same content never share a temporary file), renames into place once complete, and registers the proxy on the `IngestionContext` at that moment; the proxy middlewares only supply the bytes. If a step throws, `IngestionPipelineRunner` deletes exactly those registered proxies plus its own temporary files, and removes the directory only if it is left empty — never recursively, since a concurrent attempt on the same content shares it until one of them persists. An attempt that loses a uniqueness race (`LostIngestionRace`: the active content-hash index or the one-proxy-per-asset-and-type index) first releases its paired MOV's registrations, because the winner owns those content-derived files. A clean shutdown is treated differently: the in-flight temporary file is dropped, but proxies already completed in that run are kept, and the consumer loop re-throws the cancellation instead of recording an item failure and draining the queue. Each attempt has its own `IngestionContext.AttemptId`, which names temporary files and scopes compensation; `IngestionContext.AssetId` is the row that owns the proxies, set by classification (the stored row for `ContentChanged`, the attempt id for new content). A hard kill, the storage-config cascade delete, and the old content directory of a file edited in place are out of reach in-process and belong to the orphan sweeper (#174); the old directory is deliberately not deleted after commit, because a copy of the old content elsewhere can be ingested into it in the same crawl. ### Flashback Interaction Rules diff --git a/src/Anichron.Core.Tests.Unit/Data/Repository/MediaAssetRepositoryTests.cs b/src/Anichron.Core.Tests.Unit/Data/Repository/MediaAssetRepositoryTests.cs index d14f7c07..d36d10da 100644 --- a/src/Anichron.Core.Tests.Unit/Data/Repository/MediaAssetRepositoryTests.cs +++ b/src/Anichron.Core.Tests.Unit/Data/Repository/MediaAssetRepositoryTests.cs @@ -177,4 +177,96 @@ public async Task LoadMetadataAsync_LoadsTheAssetsMetadataRow() tracked.Metadata.Should().NotBeNull(); tracked.Metadata.Width.Should().Be(640); } + + // ========================================================================== + // FindSharedContentAsync + // ========================================================================== + + private static readonly Guid Mira = Guid.Parse("aaaaaaaa-0000-0000-0000-000000000001"); + private static readonly Guid Alice = Guid.Parse("aaaaaaaa-0000-0000-0000-000000000002"); + + private static UserStorageConfig ConfigOf(Guid userId) + => new() { Id = Guid.NewGuid(), UserId = userId, RootPath = $"/data/originals/{Guid.NewGuid():N}", IsActive = true }; + + private static async Task> FindSharedContentAsync( + UserStorageConfig[] configs, MediaAsset[] assets, IReadOnlyCollection crawled) + { + var ct = TestContext.Current.CancellationToken; + await using var seed = CreateDb(out var name); + seed.StorageConfigs.AddRange(configs); + seed.MediaAssets.AddRange(assets); + await seed.SaveChangesAsync(ct); + await using var db = AnichronDbContexts.InMemory(name); + + return await new EfMediaAssetRepository(db).FindSharedContentAsync(crawled, ct); + } + + [Fact] + public async Task FindSharedContentAsync_SameHashInTwoConfigsOfOneUser_ReturnsBothLocations() + { + var (home, phone) = (ConfigOf(Mira), ConfigOf(Mira)); + + var result = await FindSharedContentAsync( + [home, phone], + [AssetAt(home.Id, "cafe", "2023/a.jpg"), AssetAt(phone.Id, "cafe", "dcim/a.jpg"), AssetAt(home.Id, "beef", "b.jpg")], + [home.Id, phone.Id]); + + result.Should().BeEquivalentTo( + [ + new ContentLocation(Mira, "cafe", home.Id, "2023/a.jpg"), + new ContentLocation(Mira, "cafe", phone.Id, "dcim/a.jpg"), + ]); + } + + [Fact] + public async Task FindSharedContentAsync_SameHashUnderTwoUsers_IsLeftOut() + { + var (mine, theirs) = (ConfigOf(Mira), ConfigOf(Alice)); + + var result = await FindSharedContentAsync( + [mine, theirs], + [AssetAt(mine.Id, "cafe", "a.jpg"), AssetAt(theirs.Id, "cafe", "a.jpg")], + [mine.Id, theirs.Id]); + + result.Should().BeEmpty(); + } + + [Fact] + public async Task FindSharedContentAsync_ConfigNotCrawled_IsLeftOut() + { + var (home, phone) = (ConfigOf(Mira), ConfigOf(Mira)); + + var result = await FindSharedContentAsync( + [home, phone], + [AssetAt(home.Id, "cafe", "a.jpg"), AssetAt(phone.Id, "cafe", "a.jpg")], + [home.Id]); + + result.Should().BeEmpty(); + } + + [Fact] + public async Task FindSharedContentAsync_SoftDeletedCopy_IsLeftOut() + { + var (home, phone) = (ConfigOf(Mira), ConfigOf(Mira)); + + var result = await FindSharedContentAsync( + [home, phone], + [AssetAt(home.Id, "cafe", "a.jpg"), AssetAt(phone.Id, "cafe", "a.jpg", softDeleted: true)], + [home.Id, phone.Id]); + + result.Should().BeEmpty(); + } + + [Fact] + public void FindSharedContent_TranslatesToOneStatementScopedToTheCrawledConfigs() + { + using var db = AnichronDbContexts.NpgsqlModel(); + + var sql = EfMediaAssetRepository.SharedContentQuery(db, [Guid.NewGuid()]).ToQueryString(); + + sql.Should().Contain("GROUP BY") + .And.Contain("HAVING count(*)::int > 1") + .And.Contain("\"StorageConfigId\" = ANY (@") + .And.Contain("\"IsSoftDeleted\""); + } } diff --git a/src/Anichron.Core/Data/Repository/MediaAssetRepository.cs b/src/Anichron.Core/Data/Repository/MediaAssetRepository.cs index ceff94f7..9d0b92f5 100644 --- a/src/Anichron.Core/Data/Repository/MediaAssetRepository.cs +++ b/src/Anichron.Core/Data/Repository/MediaAssetRepository.cs @@ -3,6 +3,8 @@ namespace Anichron.Core.Data.Repository; +public sealed record ContentLocation(Guid UserId, string ContentHash, Guid StorageConfigId, string FilePath); + public interface IMediaAssetRepository { Task FindByHashAsync(string contentHash, Guid storageConfigId, CancellationToken ct); @@ -18,6 +20,11 @@ Task> FindByHashesOrPathsAsync( IReadOnlyCollection filePaths, CancellationToken ct); + /// Active rows whose content is also stored in another of the same user's configs. + /// Only rows in these configs are compared. + /// Cancels the query. + Task> FindSharedContentAsync(IReadOnlyCollection storageConfigIds, CancellationToken ct); + Task LoadMetadataAsync(MediaAsset asset, CancellationToken ct); void Add(MediaAsset asset); @@ -43,6 +50,23 @@ public async Task> FindByHashesOrPathsAsync( && ((!a.IsSoftDeleted && contentHashes.Contains(a.ContentHash)) || filePaths.Contains(a.FilePath))) .ToListAsync(ct); + public async Task> FindSharedContentAsync( + IReadOnlyCollection storageConfigIds, CancellationToken ct) + => await SharedContentQuery(db, storageConfigIds).ToListAsync(ct); + + internal static IQueryable SharedContentQuery( + AnichronDbContext db, IReadOnlyCollection storageConfigIds) + { + var inScope = db.MediaAssets.Where(a => storageConfigIds.Contains(a.StorageConfigId)); + var shared = inScope + .GroupBy(a => new { a.StorageConfig.UserId, a.ContentHash }) + .Where(g => g.Count() > 1) + .Select(g => g.Key); + return inScope + .Where(a => shared.Any(k => k.UserId == a.StorageConfig.UserId && k.ContentHash == a.ContentHash)) + .Select(a => new ContentLocation(a.StorageConfig.UserId, a.ContentHash, a.StorageConfigId, a.FilePath)); + } + public Task LoadMetadataAsync(MediaAsset asset, CancellationToken ct) => db.Entry(asset).Reference(a => a.Metadata).LoadAsync(ct); diff --git a/src/Anichron.Worker.Tests.Unit/Crawling/SharedContentReporterTests.cs b/src/Anichron.Worker.Tests.Unit/Crawling/SharedContentReporterTests.cs new file mode 100644 index 00000000..6709cf6c --- /dev/null +++ b/src/Anichron.Worker.Tests.Unit/Crawling/SharedContentReporterTests.cs @@ -0,0 +1,82 @@ +using Anichron.Core.Data.Repository; +using Anichron.Worker.Crawling; +using Microsoft.Extensions.Logging; +using Microsoft.Extensions.Logging.Testing; + +namespace Anichron.Worker.Tests.Unit.Crawling; + +public sealed class SharedContentReporterTests +{ + private static readonly Guid Mira = Guid.Parse("aaaaaaaa-0000-0000-0000-000000000001"); + private static readonly Guid Alice = Guid.Parse("aaaaaaaa-0000-0000-0000-000000000002"); + private static readonly Guid Home = Guid.Parse("11111111-0000-0000-0000-000000000000"); + private static readonly Guid Phone = Guid.Parse("22222222-0000-0000-0000-000000000000"); + private static readonly Guid Backup = Guid.Parse("33333333-0000-0000-0000-000000000000"); + + private static CancellationToken Ct => TestContext.Current.CancellationToken; + + private static async Task> ReportAsync( + IReadOnlyCollection crawled, params ContentLocation[] locations) + { + var repository = Substitute.For(); + repository.FindSharedContentAsync(crawled, Arg.Any()).Returns(locations); + var logger = new FakeLogger(); + + await new SharedContentReporter(repository, logger).ReportAsync(crawled, Ct); + + return logger.Collector.GetSnapshot(); + } + + [Fact] + public async Task ReportAsync_ContentInTwoConfigs_LogsThePairOnceAtInformationAsync() + { + var records = await ReportAsync( + [Home, Phone], + new ContentLocation(Mira, "cafe", Phone, "dcim/a.jpg"), + new ContentLocation(Mira, "cafe", Home, "2023/a.jpg")); + + records.Should().ContainSingle().Which.Should().Match(r => r.Level == LogLevel.Information + && r.Message == "Duplicate content cafe for user aaaaaaaa-0000-0000-0000-000000000001: " + + "2023/a.jpg in storage config 11111111-0000-0000-0000-000000000000 and " + + "dcim/a.jpg in storage config 22222222-0000-0000-0000-000000000000."); + } + + [Fact] + public async Task ReportAsync_ContentInThreeConfigs_LogsEachPairOnceAsync() + { + var records = await ReportAsync( + [Home, Phone, Backup], + new ContentLocation(Mira, "cafe", Backup, "c.jpg"), + new ContentLocation(Mira, "cafe", Home, "a.jpg"), + new ContentLocation(Mira, "cafe", Phone, "b.jpg")); + + records.Select(r => (r.GetStructuredStateValue("FilePath"), r.GetStructuredStateValue("OtherFilePath"))) + .Should().Equal(("a.jpg", "b.jpg"), ("a.jpg", "c.jpg"), ("b.jpg", "c.jpg")); + } + + [Fact] + public async Task ReportAsync_TwoHashesAndTwoUsers_PairsOnlyWithinEachUsersHashAsync() + { + var records = await ReportAsync( + [Home, Phone, Backup], + new ContentLocation(Mira, "cafe", Home, "a.jpg"), + new ContentLocation(Mira, "beef", Home, "x.jpg"), + new ContentLocation(Alice, "cafe", Backup, "theirs.jpg"), + new ContentLocation(Mira, "cafe", Phone, "b.jpg"), + new ContentLocation(Alice, "cafe", Phone, "theirs-too.jpg"), + new ContentLocation(Mira, "beef", Phone, "y.jpg")); + + records.Select(r => (r.GetStructuredStateValue("UserId"), r.GetStructuredStateValue("ContentHash"), + r.GetStructuredStateValue("FilePath"), r.GetStructuredStateValue("OtherFilePath"))) + .Should().BeEquivalentTo( + [ + (Mira.ToString(), "cafe", "a.jpg", "b.jpg"), + (Mira.ToString(), "beef", "x.jpg", "y.jpg"), + (Alice.ToString(), "cafe", "theirs-too.jpg", "theirs.jpg"), + ]); + } + + [Fact] + public async Task ReportAsync_NothingShared_LogsNothingAsync() + => (await ReportAsync([Home])).Should().BeEmpty(); +} diff --git a/src/Anichron.Worker.Tests.Unit/Crawling/WorkerTests.cs b/src/Anichron.Worker.Tests.Unit/Crawling/WorkerTests.cs index 9f6559c7..9d160041 100644 --- a/src/Anichron.Worker.Tests.Unit/Crawling/WorkerTests.cs +++ b/src/Anichron.Worker.Tests.Unit/Crawling/WorkerTests.cs @@ -23,13 +23,16 @@ private sealed class TestFixture { public IUserStorageConfigRepository ConfigRepository { get; } = Substitute.For(); public IFileIngestionPipeline Pipeline { get; } = Substitute.For(); + public ISharedContentReporter Reporter { get; } = Substitute.For(); public WorkerState WorkerState { get; } = new(); private readonly IServiceScopeFactory scopeFactory; public TestFixture() { - scopeFactory = ScopeFactories.Providing(services => services.AddSingleton(ConfigRepository)); + scopeFactory = ScopeFactories.Providing(services => services + .AddSingleton(ConfigRepository) + .AddSingleton(Reporter)); ConfigRepository.GetAllActiveAsync(Arg.Any()) .Returns([]); @@ -196,6 +199,9 @@ public async Task CrawlAllAsync_ConfigCrawlFails_LogsItsRootAndCrawlsTheNextConf await crawlingWorker.CrawlAllAsync(Ct); await fixture.Pipeline.Received(1).RunAsync(next, Arg.Any()); + await fixture.Reporter.Received(1).ReportAsync( + Arg.Is>(ids => ids.SequenceEqual(new[] { missing.Id, next.Id })), + Arg.Any()); fixture.Logger.Collector.GetSnapshot().Should().ContainSingle(r => r.Level == LogLevel.Error) .Which.Should().Match(r => r.Message == $"Crawl of /data/originals/mira (userId={userId}) failed." @@ -218,6 +224,7 @@ public async Task CrawlAllAsync_ShutdownDuringACrawl_StopsWithoutLoggingAFailure await act.Should().ThrowAsync(); await fixture.Pipeline.DidNotReceive().RunAsync(second, Arg.Any()); + await fixture.Reporter.DidNotReceive().ReportAsync(Arg.Any>(), Arg.Any()); fixture.Logger.Collector.GetSnapshot().Should().NotContain(r => r.Level == LogLevel.Error); } @@ -237,6 +244,61 @@ public async Task CrawlAllAsync_CancellationWithoutShutdown_IsLoggedAsACrawlFail .Which.Exception.Should().BeOfType(); } + // ========================================================================== + // Duplicate content report + // ========================================================================== + + [Fact] + public async Task CrawlAllAsync_AfterEveryConfigIsCrawled_ReportsSharedContentAcrossThemAsync() + { + var fixture = new TestFixture(); + var userId = Guid.NewGuid(); + var home = new UserStorageConfig { Id = Guid.NewGuid(), UserId = userId, RootPath = "/data/originals/mira" }; + var phone = new UserStorageConfig { Id = Guid.NewGuid(), UserId = userId, RootPath = "/data/originals/mira-phone" }; + fixture.WorkerState.ResolvedUserId = userId; + fixture.ConfigRepository.GetActiveByUserIdAsync(userId, Arg.Any()).Returns([home, phone]); + + await fixture.Build().CrawlAllAsync(Ct); + + Received.InOrder(() => + { + _ = fixture.Pipeline.RunAsync(home, Arg.Any()); + _ = fixture.Pipeline.RunAsync(phone, Arg.Any()); + _ = fixture.Reporter.ReportAsync( + Arg.Is>(ids => ids.SequenceEqual(new[] { home.Id, phone.Id })), + Arg.Any()); + }); + } + + [Fact] + public async Task CrawlAllAsync_ReportFails_LogsItWithoutThrowingAsync() + { + var fixture = new TestFixture(); + fixture.Reporter.ReportAsync(Arg.Any>(), Arg.Any()) + .Returns(Task.FromException(new InvalidOperationException("db down"))); + + await fixture.Build().CrawlAllAsync(Ct); + + fixture.Logger.Collector.GetSnapshot().Should().ContainSingle(r => r.Level == LogLevel.Error) + .Which.Should().Match(r => + r.Message == "Reporting duplicate content failed." && r.Exception is InvalidOperationException); + } + + [Fact] + public async Task CrawlAllAsync_ShutdownDuringTheReport_StopsWithoutLoggingAFailureAsync() + { + var fixture = new TestFixture(); + using var shutdown = new CancellationTokenSource(); + await shutdown.CancelAsync(); + fixture.Reporter.ReportAsync(Arg.Any>(), Arg.Any()) + .Returns(Task.FromCanceled(shutdown.Token)); + + var act = () => fixture.Build().CrawlAllAsync(shutdown.Token); + + await act.Should().ThrowAsync(); + fixture.Logger.Collector.GetSnapshot().Should().NotContain(r => r.Level == LogLevel.Error); + } + [Fact] public async Task ExecuteAsync_CrawlFails_KeepsRunningAndCrawlsAgainAfterTheIntervalAsync() { diff --git a/src/Anichron.Worker.Tests.Unit/Infrastructure/WorkerServiceCollectionExtensionsTests.cs b/src/Anichron.Worker.Tests.Unit/Infrastructure/WorkerServiceCollectionExtensionsTests.cs index ed479f37..af05d6c4 100644 --- a/src/Anichron.Worker.Tests.Unit/Infrastructure/WorkerServiceCollectionExtensionsTests.cs +++ b/src/Anichron.Worker.Tests.Unit/Infrastructure/WorkerServiceCollectionExtensionsTests.cs @@ -212,6 +212,7 @@ public void AddWorkerHostedServices_RegistersFourHostedServices() { { typeof(ILivePhotoLinker), typeof(LivePhotoLinker), ServiceLifetime.Singleton }, { typeof(IFileIngestionPipeline), typeof(FileIngestionPipeline), ServiceLifetime.Singleton }, + { typeof(ISharedContentReporter), typeof(SharedContentReporter), ServiceLifetime.Scoped }, { typeof(PairedMovResolver), typeof(PairedMovResolver), ServiceLifetime.Scoped }, { typeof(PairedMovProxyWriter), typeof(PairedMovProxyWriter), ServiceLifetime.Scoped }, { typeof(IngestionSaver), typeof(IngestionSaver), ServiceLifetime.Scoped }, diff --git a/src/Anichron.Worker.Tests.Unit/Ingestion/Middlewares/ClassificationMiddlewareTests.cs b/src/Anichron.Worker.Tests.Unit/Ingestion/Middlewares/ClassificationMiddlewareTests.cs index a59b13c8..90bb3e83 100644 --- a/src/Anichron.Worker.Tests.Unit/Ingestion/Middlewares/ClassificationMiddlewareTests.cs +++ b/src/Anichron.Worker.Tests.Unit/Ingestion/Middlewares/ClassificationMiddlewareTests.cs @@ -16,6 +16,7 @@ public sealed class ClassificationMiddlewareTests private static MediaAsset Stored(string filePath, string contentHash = "abc123") => new() { Id = Guid.Parse("eeeeeeee-eeee-eeee-eeee-eeeeeeeeeeee"), + StorageConfigId = ConfigId, FilePath = filePath, FileName = Path.GetFileName(filePath), ContentHash = contentHash, @@ -25,6 +26,7 @@ public sealed class ClassificationMiddlewareTests private static MediaAsset StoredMov(string filePath) => new() { Id = Guid.Parse("aaaaaaaa-aaaa-aaaa-aaaa-aaaaaaaaaaaa"), + StorageConfigId = ConfigId, FilePath = filePath, FileName = Path.GetFileName(filePath), ContentHash = "cafebabe", @@ -182,7 +184,8 @@ public async Task InvokeAsync_StoredPathStillExists_LeavesRowUnchangedAsync() stored.LastSeenOnNas.Should().Be(FirstSeen); await fx.UnitOfWork.DidNotReceive().SaveChangesAsync(Arg.Any()); fx.Logger.Collector.GetSnapshot().Should().ContainSingle() - .Which.Should().Match(r => r.Level == LogLevel.Debug && r.Message == "copy/photo.jpg is a copy of already-ingested 2023/photo.jpg, skipping."); + .Which.Should().Match(r => r.Level == LogLevel.Information && r.Message + == "Duplicate content abc123 in storage config 11111111-1111-1111-1111-111111111111: copy/photo.jpg is a copy of 2023/photo.jpg."); } [Fact] @@ -343,4 +346,22 @@ public async Task InvokeAsync_BothHalvesDuplicates_SavesNothingAsync() await fx.UnitOfWork.DidNotReceive().SaveChangesAsync(Arg.Any()); } + + [Fact] + public async Task InvokeAsync_BothHalvesDuplicates_ReportsEachHalfAsync() + { + var fx = new ClassificationFixture(); + fx.FileSystem.AddEmptyFile("/abs/2023/live.heic"); + fx.FileSystem.AddEmptyFile("/abs/2023/live.mov"); + var heic = Stored("2023/live.heic"); + heic.PairedAssetId = Guid.NewGuid(); + fx.Store(heic, StoredMov("2023/live.mov")); + + await fx.Build().InvokeAsync(Pair("copy/live.heic", "copy/live.mov"), NoOpNextAsync, CancellationToken.None); + + fx.Logger.Collector.GetSnapshot().Where(r => r.Level == LogLevel.Information).Select(r => r.Message) + .Should().Equal( + "Duplicate content abc123 in storage config 11111111-1111-1111-1111-111111111111: copy/live.heic is a copy of 2023/live.heic.", + "Duplicate content cafebabe in storage config 11111111-1111-1111-1111-111111111111: copy/live.mov is a copy of 2023/live.mov."); + } } diff --git a/src/Anichron.Worker/Crawling/SharedContentReporter.cs b/src/Anichron.Worker/Crawling/SharedContentReporter.cs new file mode 100644 index 00000000..ba5cd4cc --- /dev/null +++ b/src/Anichron.Worker/Crawling/SharedContentReporter.cs @@ -0,0 +1,40 @@ +using Anichron.Core.Data.Repository; + +namespace Anichron.Worker.Crawling; + +internal interface ISharedContentReporter +{ + Task ReportAsync(IReadOnlyCollection storageConfigIds, CancellationToken ct); +} + +internal sealed partial class SharedContentReporter( + IMediaAssetRepository repository, + ILogger logger) : ISharedContentReporter +{ + public async Task ReportAsync(IReadOnlyCollection storageConfigIds, CancellationToken ct) + { + var locations = await repository.FindSharedContentAsync(storageConfigIds, ct); + foreach (var content in locations.GroupBy(location => (location.UserId, location.ContentHash))) + { + var copies = content.OrderBy(location => location.StorageConfigId).ToList(); + var pairs = copies.SelectMany((first, index) => copies.Skip(index + 1).Select(second => (first, second))); + foreach (var (first, second) in pairs) + { + Log.SharedContent( + logger, first.ContentHash, first.UserId, + first.FilePath, first.StorageConfigId, second.FilePath, second.StorageConfigId); + } + } + } + + private static partial class Log + { + [LoggerMessage( + Level = LogLevel.Information, + Message = "Duplicate content {ContentHash} for user {UserId}: {FilePath} in storage config {StorageConfigId} " + + "and {OtherFilePath} in storage config {OtherStorageConfigId}.")] + public static partial void SharedContent( + ILogger logger, string contentHash, Guid userId, + string filePath, Guid storageConfigId, string otherFilePath, Guid otherStorageConfigId); + } +} diff --git a/src/Anichron.Worker/Crawling/Worker.cs b/src/Anichron.Worker/Crawling/Worker.cs index 6ba7fe51..e7e72a79 100644 --- a/src/Anichron.Worker/Crawling/Worker.cs +++ b/src/Anichron.Worker/Crawling/Worker.cs @@ -38,6 +38,8 @@ internal async Task CrawlAllAsync(CancellationToken ct) foreach (var config in configs) await CrawlAsync(config, ct); + + await ReportSharedContentAsync(scope.ServiceProvider, [.. configs.Select(config => config.Id)], ct); } private async Task CrawlAsync(UserStorageConfig config, CancellationToken ct) @@ -53,6 +55,19 @@ private async Task CrawlAsync(UserStorageConfig config, CancellationToken ct) } } + private async Task ReportSharedContentAsync( + IServiceProvider services, IReadOnlyCollection storageConfigIds, CancellationToken ct) + { + try + { + await services.GetRequiredService().ReportAsync(storageConfigIds, ct); + } + catch (Exception ex) when (!IngestionShutdown.IsInProgress(ex, ct)) + { + Log.SharedContentReportFailed(logger, ex); + } + } + private static partial class Log { [LoggerMessage(Level = LogLevel.Information, Message = "Worker starting in {Mode} mode.")] @@ -63,5 +78,8 @@ private static partial class Log [LoggerMessage(Level = LogLevel.Error, Message = "Crawl of {RootPath} (userId={UserId}) failed.")] public static partial void CrawlFailed(ILogger logger, string rootPath, Guid userId, Exception ex); + + [LoggerMessage(Level = LogLevel.Error, Message = "Reporting duplicate content failed.")] + public static partial void SharedContentReportFailed(ILogger logger, Exception ex); } } diff --git a/src/Anichron.Worker/Infrastructure/WorkerServiceCollectionExtensions.cs b/src/Anichron.Worker/Infrastructure/WorkerServiceCollectionExtensions.cs index b2773691..901f1e6f 100644 --- a/src/Anichron.Worker/Infrastructure/WorkerServiceCollectionExtensions.cs +++ b/src/Anichron.Worker/Infrastructure/WorkerServiceCollectionExtensions.cs @@ -55,6 +55,7 @@ internal IServiceCollection AddIngestionServices() services.AddSingleton(); services.AddSingleton(); services.AddSingleton(); + services.AddScoped(); services.AddSingleton(); services.AddSingleton(); services.AddImageProxyServices(); diff --git a/src/Anichron.Worker/Ingestion/Classification/KnownFileReconciler.cs b/src/Anichron.Worker/Ingestion/Classification/KnownFileReconciler.cs index fb910e5c..256a4886 100644 --- a/src/Anichron.Worker/Ingestion/Classification/KnownFileReconciler.cs +++ b/src/Anichron.Worker/Ingestion/Classification/KnownFileReconciler.cs @@ -32,7 +32,7 @@ private bool Move(MediaAsset stored, string relativePath) private bool SkipDuplicate(MediaAsset stored, string relativePath) { - Log.DuplicateCopy(logger, relativePath, stored.FilePath); + Log.DuplicateCopy(logger, stored.ContentHash, stored.StorageConfigId, relativePath, stored.FilePath); return false; } @@ -47,8 +47,11 @@ private static partial class Log [LoggerMessage(Level = LogLevel.Debug, Message = "Already ingested {RelativePath}, refreshed last seen.")] public static partial void AlreadyIngested(ILogger logger, string relativePath); - [LoggerMessage(Level = LogLevel.Debug, Message = "{RelativePath} is a copy of already-ingested {StoredPath}, skipping.")] - public static partial void DuplicateCopy(ILogger logger, string relativePath, string storedPath); + [LoggerMessage( + Level = LogLevel.Information, + Message = "Duplicate content {ContentHash} in storage config {StorageConfigId}: {RelativePath} is a copy of {StoredPath}.")] + public static partial void DuplicateCopy( + ILogger logger, string contentHash, Guid storageConfigId, string relativePath, string storedPath); [LoggerMessage(Level = LogLevel.Information, Message = "Asset {AssetId} moved from {OldPath} to {NewPath}")] public static partial void Moved(ILogger logger, Guid assetId, string oldPath, string newPath); From f33de2de065e201e632b1ca450945bc8657de6d3 Mon Sep 17 00:00:00 2001 From: bitbiter-dev Date: Mon, 19 Oct 2026 14:03:16 +1300 Subject: [PATCH 2/3] docs: Say the duplicate report repeats every crawl cycle (#36) Co-Authored-By: Claude Opus 5.5 --- CLAUDE.md | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/CLAUDE.md b/CLAUDE.md index 938b6a2b..145075db 100644 --- a/CLAUDE.md +++ b/CLAUDE.md @@ -157,7 +157,7 @@ The API reads files only through two narrow stores in `Anichron.Infrastructure/S - FFmpeg transcodes video with runtime GPU detection: QuickSync (`h264_qsv`) → NVENC (`h264_nvenc`) → AMF (`h264_amf`) → software (`libx264`) - Burst detection (**planned, not yet implemented**): will group rapid-fire sequences and assign a `PrimaryAssetId` cover. The `Burst` entity and schema exist; no detection code is wired into the Worker yet. - Reconciliation (**planned, Epic 4**): a periodic scan soft-deletes assets whose file is gone (interactions survive) and restores them when it reappears (#33, #34, #37). Moves are already handled at ingest (#35): when the hash matches an active row, `KnownFileReconciler` refreshes `LastSeenOnNas` if the path is unchanged (`Unchanged`), moves the row to the new path if its stored file is gone (`Moved`), and leaves it alone if the stored file still exists (`DuplicateCopy`, logged at `Information` with hash, config and both paths — #36). A Live Photo's MOV row gets the same treatment by its own hash. A move onto a path another row already holds, soft-deleted rows included (`PathTaken`), is logged as a warning and skipped; if a concurrent consumer takes the path after classification, the save fails and the item is retried on the next crawl. A file **edited in place** (hash unknown, but an active row holds the path: `ContentChanged`) is processed like new content and `ContentChangedWriter` updates that row: new hash, capture date, media type, `Metadata` and proxy rows (updated by type), while `Id`, interactions, `BurstId` and `PairedAssetId` are kept. A soft-deleted row at the path still blocks it (restore, #34). An edited Live Photo **MOV** keeps its stored row unchanged for now -- Duplicate content is **reported, never merged** (#36): besides `DuplicateCopy` within a config, `SharedContentReporter` runs after each crawl cycle and logs, once per pair, active rows of one user that share a `ContentHash` across the configs just crawled. Duplicates across different users are expected and never reported. Two new identical copies racing in one crawl are first reported on the next crawl +- Duplicate content is **reported, never merged** (#36): besides `DuplicateCopy` within a config, `SharedContentReporter` runs after each crawl cycle and logs, once per pair per cycle, active rows of one user that share a `ContentHash` across the configs just crawled. Duplicates across different users are expected and never reported. Two new identical copies racing in one crawl are first reported on the next crawl - Failure ownership: `ProxyStagingWriter` is the single place a proxy reaches its final name (architecture rule `proxies-reach-final-name-only-in-staging-writer`) — it writes under a per-attempt `..tmp` suffix (so concurrent attempts on the same content never share a temporary file), renames into place once complete, and registers the proxy on the `IngestionContext` at that moment; the proxy middlewares only supply the bytes. If a step throws, `IngestionPipelineRunner` deletes exactly those registered proxies plus its own temporary files, and removes the directory only if it is left empty — never recursively, since a concurrent attempt on the same content shares it until one of them persists. An attempt that loses a uniqueness race (`LostIngestionRace`: the active content-hash index or the one-proxy-per-asset-and-type index) first releases its paired MOV's registrations, because the winner owns those content-derived files. A clean shutdown is treated differently: the in-flight temporary file is dropped, but proxies already completed in that run are kept, and the consumer loop re-throws the cancellation instead of recording an item failure and draining the queue. Each attempt has its own `IngestionContext.AttemptId`, which names temporary files and scopes compensation; `IngestionContext.AssetId` is the row that owns the proxies, set by classification (the stored row for `ContentChanged`, the attempt id for new content). A hard kill, the storage-config cascade delete, and the old content directory of a file edited in place are out of reach in-process and belong to the orphan sweeper (#174); the old directory is deliberately not deleted after commit, because a copy of the old content elsewhere can be ingested into it in the same crawl. ### Flashback Interaction Rules From 760ba8aa749451d7eaa162745608ace7a0f6ddf1 Mon Sep 17 00:00:00 2001 From: bitbiter-dev Date: Mon, 19 Oct 2026 14:19:27 +1300 Subject: [PATCH 3/3] feat(worker): One duplicate summary per user per cycle, details at Debug (#36) A library with a mirrored config produced one Information line per duplicate on every crawl cycle. Each copy and pair is now logged at Debug, and the end-of-cycle report logs one Information summary per user: copies within one storage config, counted by DuplicateCopyTally while the crawl runs, and pairs across storage configs. Co-Authored-By: Claude Opus 5.5 --- CLAUDE.md | 4 +- .../Crawling/SharedContentReporterTests.cs | 114 ++++++++++++++---- .../Crawling/WorkerTests.cs | 10 +- .../WorkerServiceCollectionExtensionsTests.cs | 1 + .../Classification/DuplicateCopyTallyTests.cs | 41 +++++++ .../Middlewares/AlwaysOnMiddlewareTests.cs | 2 +- .../Middlewares/ClassificationFixture.cs | 3 +- .../ClassificationMiddlewareTests.cs | 6 +- .../Crawling/SharedContentReporter.cs | 31 ++++- src/Anichron.Worker/Crawling/Worker.cs | 6 +- .../WorkerServiceCollectionExtensions.cs | 1 + .../Classification/DuplicateCopyTally.cs | 12 ++ .../Classification/KnownFileReconciler.cs | 8 +- 13 files changed, 197 insertions(+), 42 deletions(-) create mode 100644 src/Anichron.Worker.Tests.Unit/Ingestion/Classification/DuplicateCopyTallyTests.cs create mode 100644 src/Anichron.Worker/Ingestion/Classification/DuplicateCopyTally.cs diff --git a/CLAUDE.md b/CLAUDE.md index 145075db..01763447 100644 --- a/CLAUDE.md +++ b/CLAUDE.md @@ -156,8 +156,8 @@ The API reads files only through two narrow stores in `Anichron.Infrastructure/S - Each `IImageProxyGenerator`, and the `BlurhashEncoder`, declares a `ProxyBound`; `ImageProxyMiddleware` decodes once to `Rgb24`, shrinks the shared image to each bound largest-first, and hands every step **its own copy** at that size. The encoder's string goes on the `IngestionContext` and the content writer (`ProcessedContentWriter.DescribeContent`) writes it to `Metadata`. Generators may crop/recolour/resize their copy freely; only the middleware resizes the shared image - FFmpeg transcodes video with runtime GPU detection: QuickSync (`h264_qsv`) → NVENC (`h264_nvenc`) → AMF (`h264_amf`) → software (`libx264`) - Burst detection (**planned, not yet implemented**): will group rapid-fire sequences and assign a `PrimaryAssetId` cover. The `Burst` entity and schema exist; no detection code is wired into the Worker yet. -- Reconciliation (**planned, Epic 4**): a periodic scan soft-deletes assets whose file is gone (interactions survive) and restores them when it reappears (#33, #34, #37). Moves are already handled at ingest (#35): when the hash matches an active row, `KnownFileReconciler` refreshes `LastSeenOnNas` if the path is unchanged (`Unchanged`), moves the row to the new path if its stored file is gone (`Moved`), and leaves it alone if the stored file still exists (`DuplicateCopy`, logged at `Information` with hash, config and both paths — #36). A Live Photo's MOV row gets the same treatment by its own hash. A move onto a path another row already holds, soft-deleted rows included (`PathTaken`), is logged as a warning and skipped; if a concurrent consumer takes the path after classification, the save fails and the item is retried on the next crawl. A file **edited in place** (hash unknown, but an active row holds the path: `ContentChanged`) is processed like new content and `ContentChangedWriter` updates that row: new hash, capture date, media type, `Metadata` and proxy rows (updated by type), while `Id`, interactions, `BurstId` and `PairedAssetId` are kept. A soft-deleted row at the path still blocks it (restore, #34). An edited Live Photo **MOV** keeps its stored row unchanged for now -- Duplicate content is **reported, never merged** (#36): besides `DuplicateCopy` within a config, `SharedContentReporter` runs after each crawl cycle and logs, once per pair per cycle, active rows of one user that share a `ContentHash` across the configs just crawled. Duplicates across different users are expected and never reported. Two new identical copies racing in one crawl are first reported on the next crawl +- Reconciliation (**planned, Epic 4**): a periodic scan soft-deletes assets whose file is gone (interactions survive) and restores them when it reappears (#33, #34, #37). Moves are already handled at ingest (#35): when the hash matches an active row, `KnownFileReconciler` refreshes `LastSeenOnNas` if the path is unchanged (`Unchanged`), moves the row to the new path if its stored file is gone (`Moved`), and leaves it alone if the stored file still exists (`DuplicateCopy`, reported by #36). A Live Photo's MOV row gets the same treatment by its own hash. A move onto a path another row already holds, soft-deleted rows included (`PathTaken`), is logged as a warning and skipped; if a concurrent consumer takes the path after classification, the save fails and the item is retried on the next crawl. A file **edited in place** (hash unknown, but an active row holds the path: `ContentChanged`) is processed like new content and `ContentChangedWriter` updates that row: new hash, capture date, media type, `Metadata` and proxy rows (updated by type), while `Id`, interactions, `BurstId` and `PairedAssetId` are kept. A soft-deleted row at the path still blocks it (restore, #34). An edited Live Photo **MOV** keeps its stored row unchanged for now +- Duplicate content is **reported, never merged** (#36). `SharedContentReporter` runs after each crawl cycle over the configs just crawled: it logs one `Information` summary per user (copies within one config, counted by `DuplicateCopyTally` during the crawl, and pairs of that user's active rows sharing a `ContentHash` across configs), with each copy and pair at `Debug`. Duplicates across different users are expected and never reported. Two new identical copies racing in one crawl are first reported on the next crawl - Failure ownership: `ProxyStagingWriter` is the single place a proxy reaches its final name (architecture rule `proxies-reach-final-name-only-in-staging-writer`) — it writes under a per-attempt `..tmp` suffix (so concurrent attempts on the same content never share a temporary file), renames into place once complete, and registers the proxy on the `IngestionContext` at that moment; the proxy middlewares only supply the bytes. If a step throws, `IngestionPipelineRunner` deletes exactly those registered proxies plus its own temporary files, and removes the directory only if it is left empty — never recursively, since a concurrent attempt on the same content shares it until one of them persists. An attempt that loses a uniqueness race (`LostIngestionRace`: the active content-hash index or the one-proxy-per-asset-and-type index) first releases its paired MOV's registrations, because the winner owns those content-derived files. A clean shutdown is treated differently: the in-flight temporary file is dropped, but proxies already completed in that run are kept, and the consumer loop re-throws the cancellation instead of recording an item failure and draining the queue. Each attempt has its own `IngestionContext.AttemptId`, which names temporary files and scopes compensation; `IngestionContext.AssetId` is the row that owns the proxies, set by classification (the stored row for `ContentChanged`, the attempt id for new content). A hard kill, the storage-config cascade delete, and the old content directory of a file edited in place are out of reach in-process and belong to the orphan sweeper (#174); the old directory is deliberately not deleted after commit, because a copy of the old content elsewhere can be ingested into it in the same crawl. ### Flashback Interaction Rules diff --git a/src/Anichron.Worker.Tests.Unit/Crawling/SharedContentReporterTests.cs b/src/Anichron.Worker.Tests.Unit/Crawling/SharedContentReporterTests.cs index 6709cf6c..2fb21465 100644 --- a/src/Anichron.Worker.Tests.Unit/Crawling/SharedContentReporterTests.cs +++ b/src/Anichron.Worker.Tests.Unit/Crawling/SharedContentReporterTests.cs @@ -1,5 +1,7 @@ using Anichron.Core.Data.Repository; +using Anichron.Core.Domain; using Anichron.Worker.Crawling; +using Anichron.Worker.Ingestion.Classification; using Microsoft.Extensions.Logging; using Microsoft.Extensions.Logging.Testing; @@ -15,50 +17,70 @@ public sealed class SharedContentReporterTests private static CancellationToken Ct => TestContext.Current.CancellationToken; - private static async Task> ReportAsync( - IReadOnlyCollection crawled, params ContentLocation[] locations) + private static UserStorageConfig Config(Guid id, Guid userId) => new() { Id = id, UserId = userId, RootPath = "/unused" }; + + private sealed class TestFixture { - var repository = Substitute.For(); - repository.FindSharedContentAsync(crawled, Arg.Any()).Returns(locations); - var logger = new FakeLogger(); + public IMediaAssetRepository Repository { get; } = Substitute.For(); + public DuplicateCopyTally Tally { get; } = new(); + public FakeLogger Logger { get; } = new(); + + public async Task> ReportAsync( + UserStorageConfig[] crawled, params ContentLocation[] locations) + { + Repository.FindSharedContentAsync( + Arg.Is>(ids => ids.SequenceEqual(crawled.Select(c => c.Id))), + Arg.Any()) + .Returns(locations); - await new SharedContentReporter(repository, logger).ReportAsync(crawled, Ct); + await new SharedContentReporter(Repository, Tally, Logger).ReportAsync(crawled, Ct); - return logger.Collector.GetSnapshot(); + return Logger.Collector.GetSnapshot(); + } } + private static IEnumerable Pairs(IEnumerable records) + => records.Where(r => r.Level == LogLevel.Debug); + + private static IEnumerable Summaries(IEnumerable records) + => records.Where(r => r.Level == LogLevel.Information).Select(r => r.Message); + + // ========================================================================== + // Pairs across configs + // ========================================================================== + [Fact] - public async Task ReportAsync_ContentInTwoConfigs_LogsThePairOnceAtInformationAsync() + public async Task ReportAsync_ContentInTwoConfigs_LogsThePairOnceAtDebugAsync() { - var records = await ReportAsync( - [Home, Phone], + var records = await new TestFixture().ReportAsync( + [Config(Home, Mira), Config(Phone, Mira)], new ContentLocation(Mira, "cafe", Phone, "dcim/a.jpg"), new ContentLocation(Mira, "cafe", Home, "2023/a.jpg")); - records.Should().ContainSingle().Which.Should().Match(r => r.Level == LogLevel.Information - && r.Message == "Duplicate content cafe for user aaaaaaaa-0000-0000-0000-000000000001: " - + "2023/a.jpg in storage config 11111111-0000-0000-0000-000000000000 and " - + "dcim/a.jpg in storage config 22222222-0000-0000-0000-000000000000."); + Pairs(records).Should().ContainSingle().Which.Message.Should().Be( + "Duplicate content cafe for user aaaaaaaa-0000-0000-0000-000000000001: " + + "2023/a.jpg in storage config 11111111-0000-0000-0000-000000000000 and " + + "dcim/a.jpg in storage config 22222222-0000-0000-0000-000000000000."); } [Fact] public async Task ReportAsync_ContentInThreeConfigs_LogsEachPairOnceAsync() { - var records = await ReportAsync( - [Home, Phone, Backup], + var records = await new TestFixture().ReportAsync( + [Config(Home, Mira), Config(Phone, Mira), Config(Backup, Mira)], new ContentLocation(Mira, "cafe", Backup, "c.jpg"), new ContentLocation(Mira, "cafe", Home, "a.jpg"), new ContentLocation(Mira, "cafe", Phone, "b.jpg")); - records.Select(r => (r.GetStructuredStateValue("FilePath"), r.GetStructuredStateValue("OtherFilePath"))) + Pairs(records).Select(r => (r.GetStructuredStateValue("FilePath"), r.GetStructuredStateValue("OtherFilePath"))) .Should().Equal(("a.jpg", "b.jpg"), ("a.jpg", "c.jpg"), ("b.jpg", "c.jpg")); } [Fact] public async Task ReportAsync_TwoHashesAndTwoUsers_PairsOnlyWithinEachUsersHashAsync() { - var records = await ReportAsync( - [Home, Phone, Backup], + var records = await new TestFixture().ReportAsync( + [Config(Home, Mira), Config(Phone, Mira), Config(Backup, Alice)], new ContentLocation(Mira, "cafe", Home, "a.jpg"), new ContentLocation(Mira, "beef", Home, "x.jpg"), new ContentLocation(Alice, "cafe", Backup, "theirs.jpg"), @@ -66,7 +88,7 @@ public async Task ReportAsync_TwoHashesAndTwoUsers_PairsOnlyWithinEachUsersHashA new ContentLocation(Alice, "cafe", Phone, "theirs-too.jpg"), new ContentLocation(Mira, "beef", Phone, "y.jpg")); - records.Select(r => (r.GetStructuredStateValue("UserId"), r.GetStructuredStateValue("ContentHash"), + Pairs(records).Select(r => (r.GetStructuredStateValue("UserId"), r.GetStructuredStateValue("ContentHash"), r.GetStructuredStateValue("FilePath"), r.GetStructuredStateValue("OtherFilePath"))) .Should().BeEquivalentTo( [ @@ -76,7 +98,55 @@ public async Task ReportAsync_TwoHashesAndTwoUsers_PairsOnlyWithinEachUsersHashA ]); } + // ========================================================================== + // Summary per user + // ========================================================================== + + [Fact] + public async Task ReportAsync_CopiesWithinAndAcrossConfigs_SummarisesEachUserOnceAtInformationAsync() + { + var fx = new TestFixture(); + fx.Tally.Count(Home); + fx.Tally.Count(Home); + fx.Tally.Count(Phone); + fx.Tally.Count(Backup); + + var records = await fx.ReportAsync( + [Config(Home, Mira), Config(Phone, Mira), Config(Backup, Alice)], + new ContentLocation(Mira, "cafe", Home, "a.jpg"), + new ContentLocation(Mira, "cafe", Phone, "b.jpg")); + + Summaries(records).Should().Equal( + "Duplicate content for user aaaaaaaa-0000-0000-0000-000000000001: " + + "copies within one storage config: 3, pairs across storage configs: 1.", + "Duplicate content for user aaaaaaaa-0000-0000-0000-000000000002: " + + "copies within one storage config: 1, pairs across storage configs: 0."); + } + + [Fact] + public async Task ReportAsync_OnlyAPairAcrossConfigs_StillSummarisesTheUserAsync() + { + var records = await new TestFixture().ReportAsync( + [Config(Home, Mira), Config(Phone, Mira)], + new ContentLocation(Mira, "cafe", Home, "a.jpg"), + new ContentLocation(Mira, "cafe", Phone, "b.jpg")); + + Summaries(records).Should().ContainSingle().Which.Should().EndWith( + "copies within one storage config: 0, pairs across storage configs: 1."); + } + + [Fact] + public async Task ReportAsync_Reported_StartsTheNextCycleFromZeroAsync() + { + var fx = new TestFixture(); + fx.Tally.Count(Home); + + await fx.ReportAsync([Config(Home, Mira)]); + + fx.Tally.Take(Home).Should().Be(0); + } + [Fact] - public async Task ReportAsync_NothingShared_LogsNothingAsync() - => (await ReportAsync([Home])).Should().BeEmpty(); + public async Task ReportAsync_NothingDuplicated_LogsNothingAsync() + => (await new TestFixture().ReportAsync([Config(Home, Mira), Config(Backup, Alice)])).Should().BeEmpty(); } diff --git a/src/Anichron.Worker.Tests.Unit/Crawling/WorkerTests.cs b/src/Anichron.Worker.Tests.Unit/Crawling/WorkerTests.cs index 9d160041..6d1bdee6 100644 --- a/src/Anichron.Worker.Tests.Unit/Crawling/WorkerTests.cs +++ b/src/Anichron.Worker.Tests.Unit/Crawling/WorkerTests.cs @@ -200,7 +200,7 @@ public async Task CrawlAllAsync_ConfigCrawlFails_LogsItsRootAndCrawlsTheNextConf await fixture.Pipeline.Received(1).RunAsync(next, Arg.Any()); await fixture.Reporter.Received(1).ReportAsync( - Arg.Is>(ids => ids.SequenceEqual(new[] { missing.Id, next.Id })), + Arg.Is>(crawled => crawled.SequenceEqual(new[] { missing, next })), Arg.Any()); fixture.Logger.Collector.GetSnapshot().Should().ContainSingle(r => r.Level == LogLevel.Error) .Which.Should().Match(r => @@ -224,7 +224,7 @@ public async Task CrawlAllAsync_ShutdownDuringACrawl_StopsWithoutLoggingAFailure await act.Should().ThrowAsync(); await fixture.Pipeline.DidNotReceive().RunAsync(second, Arg.Any()); - await fixture.Reporter.DidNotReceive().ReportAsync(Arg.Any>(), Arg.Any()); + await fixture.Reporter.DidNotReceive().ReportAsync(Arg.Any>(), Arg.Any()); fixture.Logger.Collector.GetSnapshot().Should().NotContain(r => r.Level == LogLevel.Error); } @@ -265,7 +265,7 @@ public async Task CrawlAllAsync_AfterEveryConfigIsCrawled_ReportsSharedContentAc _ = fixture.Pipeline.RunAsync(home, Arg.Any()); _ = fixture.Pipeline.RunAsync(phone, Arg.Any()); _ = fixture.Reporter.ReportAsync( - Arg.Is>(ids => ids.SequenceEqual(new[] { home.Id, phone.Id })), + Arg.Is>(crawled => crawled.SequenceEqual(new[] { home, phone })), Arg.Any()); }); } @@ -274,7 +274,7 @@ public async Task CrawlAllAsync_AfterEveryConfigIsCrawled_ReportsSharedContentAc public async Task CrawlAllAsync_ReportFails_LogsItWithoutThrowingAsync() { var fixture = new TestFixture(); - fixture.Reporter.ReportAsync(Arg.Any>(), Arg.Any()) + fixture.Reporter.ReportAsync(Arg.Any>(), Arg.Any()) .Returns(Task.FromException(new InvalidOperationException("db down"))); await fixture.Build().CrawlAllAsync(Ct); @@ -290,7 +290,7 @@ public async Task CrawlAllAsync_ShutdownDuringTheReport_StopsWithoutLoggingAFail var fixture = new TestFixture(); using var shutdown = new CancellationTokenSource(); await shutdown.CancelAsync(); - fixture.Reporter.ReportAsync(Arg.Any>(), Arg.Any()) + fixture.Reporter.ReportAsync(Arg.Any>(), Arg.Any()) .Returns(Task.FromCanceled(shutdown.Token)); var act = () => fixture.Build().CrawlAllAsync(shutdown.Token); diff --git a/src/Anichron.Worker.Tests.Unit/Infrastructure/WorkerServiceCollectionExtensionsTests.cs b/src/Anichron.Worker.Tests.Unit/Infrastructure/WorkerServiceCollectionExtensionsTests.cs index af05d6c4..78ce5404 100644 --- a/src/Anichron.Worker.Tests.Unit/Infrastructure/WorkerServiceCollectionExtensionsTests.cs +++ b/src/Anichron.Worker.Tests.Unit/Infrastructure/WorkerServiceCollectionExtensionsTests.cs @@ -213,6 +213,7 @@ public void AddWorkerHostedServices_RegistersFourHostedServices() { typeof(ILivePhotoLinker), typeof(LivePhotoLinker), ServiceLifetime.Singleton }, { typeof(IFileIngestionPipeline), typeof(FileIngestionPipeline), ServiceLifetime.Singleton }, { typeof(ISharedContentReporter), typeof(SharedContentReporter), ServiceLifetime.Scoped }, + { typeof(DuplicateCopyTally), typeof(DuplicateCopyTally), ServiceLifetime.Singleton }, { typeof(PairedMovResolver), typeof(PairedMovResolver), ServiceLifetime.Scoped }, { typeof(PairedMovProxyWriter), typeof(PairedMovProxyWriter), ServiceLifetime.Scoped }, { typeof(IngestionSaver), typeof(IngestionSaver), ServiceLifetime.Scoped }, diff --git a/src/Anichron.Worker.Tests.Unit/Ingestion/Classification/DuplicateCopyTallyTests.cs b/src/Anichron.Worker.Tests.Unit/Ingestion/Classification/DuplicateCopyTallyTests.cs new file mode 100644 index 00000000..e5873a4c --- /dev/null +++ b/src/Anichron.Worker.Tests.Unit/Ingestion/Classification/DuplicateCopyTallyTests.cs @@ -0,0 +1,41 @@ +using Anichron.Worker.Ingestion.Classification; + +namespace Anichron.Worker.Tests.Unit.Ingestion.Classification; + +public sealed class DuplicateCopyTallyTests +{ + private static readonly Guid Home = Guid.Parse("11111111-0000-0000-0000-000000000000"); + private static readonly Guid Phone = Guid.Parse("22222222-0000-0000-0000-000000000000"); + + [Fact] + public void Take_AfterCopiesCounted_ReturnsTheirNumberForThatConfigOnly() + { + var tally = new DuplicateCopyTally(); + tally.Count(Home); + tally.Count(Home); + tally.Count(Phone); + + tally.Take(Home).Should().Be(2); + tally.Take(Phone).Should().Be(1); + } + + [Fact] + public void Take_Twice_StartsTheNextCycleFromZero() + { + var tally = new DuplicateCopyTally(); + tally.Count(Home); + tally.Take(Home); + + tally.Take(Home).Should().Be(0); + } + + [Fact] + public void Count_FromConcurrentConsumers_LosesNoCopy() + { + var tally = new DuplicateCopyTally(); + + Parallel.For(0, 1000, new ParallelOptions { CancellationToken = TestContext.Current.CancellationToken }, _ => tally.Count(Home)); + + tally.Take(Home).Should().Be(1000); + } +} diff --git a/src/Anichron.Worker.Tests.Unit/Ingestion/Middlewares/AlwaysOnMiddlewareTests.cs b/src/Anichron.Worker.Tests.Unit/Ingestion/Middlewares/AlwaysOnMiddlewareTests.cs index 03feaaa6..0ad92b79 100644 --- a/src/Anichron.Worker.Tests.Unit/Ingestion/Middlewares/AlwaysOnMiddlewareTests.cs +++ b/src/Anichron.Worker.Tests.Unit/Ingestion/Middlewares/AlwaysOnMiddlewareTests.cs @@ -24,7 +24,7 @@ private static IIngestionMiddleware Create(string name) "Classification" => new ClassificationMiddleware( new IngestionClassifier(repository, fileSystem), new LivePhotoPairer(null!, null!, saver, new FakeLogger()), - new KnownFileReconciler(Substitute.For(), new FakeLogger()), + new KnownFileReconciler(Substitute.For(), new DuplicateCopyTally(), new FakeLogger()), saver), "ContentHashing" => new ContentHashingMiddleware(fileSystem), "Logging" => new LoggingMiddleware(new FakeLogger()), diff --git a/src/Anichron.Worker.Tests.Unit/Ingestion/Middlewares/ClassificationFixture.cs b/src/Anichron.Worker.Tests.Unit/Ingestion/Middlewares/ClassificationFixture.cs index 38522216..1f8ed933 100644 --- a/src/Anichron.Worker.Tests.Unit/Ingestion/Middlewares/ClassificationFixture.cs +++ b/src/Anichron.Worker.Tests.Unit/Ingestion/Middlewares/ClassificationFixture.cs @@ -26,6 +26,7 @@ internal sealed class ClassificationFixture public IGuidFactory GuidFactory { get; } = Substitute.For(); public IVideoProxyGenerator MovGenerator { get; } = Substitute.For(); public FakeLogger Logger { get; } = new(); + public DuplicateCopyTally Tally { get; } = new(); public List AddedProxies { get; } = []; public ClassificationFixture() @@ -56,7 +57,7 @@ public ClassificationMiddleware Build() new FakeLogger()), saver, new FakeLogger()); - return new(new IngestionClassifier(Repository, FileSystem), pairer, new KnownFileReconciler(Clock, Logger), saver); + return new(new IngestionClassifier(Repository, FileSystem), pairer, new KnownFileReconciler(Clock, Tally, Logger), saver); } public Task DidNotTranscodeAsync() diff --git a/src/Anichron.Worker.Tests.Unit/Ingestion/Middlewares/ClassificationMiddlewareTests.cs b/src/Anichron.Worker.Tests.Unit/Ingestion/Middlewares/ClassificationMiddlewareTests.cs index 90bb3e83..09686fef 100644 --- a/src/Anichron.Worker.Tests.Unit/Ingestion/Middlewares/ClassificationMiddlewareTests.cs +++ b/src/Anichron.Worker.Tests.Unit/Ingestion/Middlewares/ClassificationMiddlewareTests.cs @@ -184,8 +184,9 @@ public async Task InvokeAsync_StoredPathStillExists_LeavesRowUnchangedAsync() stored.LastSeenOnNas.Should().Be(FirstSeen); await fx.UnitOfWork.DidNotReceive().SaveChangesAsync(Arg.Any()); fx.Logger.Collector.GetSnapshot().Should().ContainSingle() - .Which.Should().Match(r => r.Level == LogLevel.Information && r.Message + .Which.Should().Match(r => r.Level == LogLevel.Debug && r.Message == "Duplicate content abc123 in storage config 11111111-1111-1111-1111-111111111111: copy/photo.jpg is a copy of 2023/photo.jpg."); + fx.Tally.Take(ConfigId).Should().Be(1); } [Fact] @@ -359,7 +360,8 @@ public async Task InvokeAsync_BothHalvesDuplicates_ReportsEachHalfAsync() await fx.Build().InvokeAsync(Pair("copy/live.heic", "copy/live.mov"), NoOpNextAsync, CancellationToken.None); - fx.Logger.Collector.GetSnapshot().Where(r => r.Level == LogLevel.Information).Select(r => r.Message) + fx.Tally.Take(ConfigId).Should().Be(2); + fx.Logger.Collector.GetSnapshot().Select(r => r.Message) .Should().Equal( "Duplicate content abc123 in storage config 11111111-1111-1111-1111-111111111111: copy/live.heic is a copy of 2023/live.heic.", "Duplicate content cafebabe in storage config 11111111-1111-1111-1111-111111111111: copy/live.mov is a copy of 2023/live.mov."); diff --git a/src/Anichron.Worker/Crawling/SharedContentReporter.cs b/src/Anichron.Worker/Crawling/SharedContentReporter.cs index ba5cd4cc..66cce671 100644 --- a/src/Anichron.Worker/Crawling/SharedContentReporter.cs +++ b/src/Anichron.Worker/Crawling/SharedContentReporter.cs @@ -1,19 +1,28 @@ using Anichron.Core.Data.Repository; +using Anichron.Core.Domain; +using Anichron.Worker.Ingestion.Classification; namespace Anichron.Worker.Crawling; internal interface ISharedContentReporter { - Task ReportAsync(IReadOnlyCollection storageConfigIds, CancellationToken ct); + Task ReportAsync(IReadOnlyCollection configs, CancellationToken ct); } internal sealed partial class SharedContentReporter( IMediaAssetRepository repository, + DuplicateCopyTally tally, ILogger logger) : ISharedContentReporter { - public async Task ReportAsync(IReadOnlyCollection storageConfigIds, CancellationToken ct) + public async Task ReportAsync(IReadOnlyCollection configs, CancellationToken ct) { - var locations = await repository.FindSharedContentAsync(storageConfigIds, ct); + var copiesByUser = configs + .GroupBy(config => config.UserId) + .Select(user => (UserId: user.Key, Copies: user.Sum(config => tally.Take(config.Id)))) + .ToList(); + + var locations = await repository.FindSharedContentAsync([.. configs.Select(config => config.Id)], ct); + var pairsByUser = new Dictionary(); foreach (var content in locations.GroupBy(location => (location.UserId, location.ContentHash))) { var copies = content.OrderBy(location => location.StorageConfigId).ToList(); @@ -23,18 +32,32 @@ public async Task ReportAsync(IReadOnlyCollection storageConfigIds, Cancel Log.SharedContent( logger, first.ContentHash, first.UserId, first.FilePath, first.StorageConfigId, second.FilePath, second.StorageConfigId); + pairsByUser[first.UserId] = pairsByUser.GetValueOrDefault(first.UserId) + 1; } } + + foreach (var (userId, copies) in copiesByUser) + { + var pairs = pairsByUser.GetValueOrDefault(userId); + if (copies + pairs > 0) + Log.Summary(logger, userId, copies, pairs); + } } private static partial class Log { [LoggerMessage( - Level = LogLevel.Information, + Level = LogLevel.Debug, Message = "Duplicate content {ContentHash} for user {UserId}: {FilePath} in storage config {StorageConfigId} " + "and {OtherFilePath} in storage config {OtherStorageConfigId}.")] public static partial void SharedContent( ILogger logger, string contentHash, Guid userId, string filePath, Guid storageConfigId, string otherFilePath, Guid otherStorageConfigId); + + [LoggerMessage( + Level = LogLevel.Information, + Message = "Duplicate content for user {UserId}: copies within one storage config: {WithinConfigCopies}, " + + "pairs across storage configs: {AcrossConfigPairs}.")] + public static partial void Summary(ILogger logger, Guid userId, int withinConfigCopies, int acrossConfigPairs); } } diff --git a/src/Anichron.Worker/Crawling/Worker.cs b/src/Anichron.Worker/Crawling/Worker.cs index e7e72a79..a734394e 100644 --- a/src/Anichron.Worker/Crawling/Worker.cs +++ b/src/Anichron.Worker/Crawling/Worker.cs @@ -39,7 +39,7 @@ internal async Task CrawlAllAsync(CancellationToken ct) foreach (var config in configs) await CrawlAsync(config, ct); - await ReportSharedContentAsync(scope.ServiceProvider, [.. configs.Select(config => config.Id)], ct); + await ReportSharedContentAsync(scope.ServiceProvider, configs, ct); } private async Task CrawlAsync(UserStorageConfig config, CancellationToken ct) @@ -56,11 +56,11 @@ private async Task CrawlAsync(UserStorageConfig config, CancellationToken ct) } private async Task ReportSharedContentAsync( - IServiceProvider services, IReadOnlyCollection storageConfigIds, CancellationToken ct) + IServiceProvider services, IReadOnlyCollection configs, CancellationToken ct) { try { - await services.GetRequiredService().ReportAsync(storageConfigIds, ct); + await services.GetRequiredService().ReportAsync(configs, ct); } catch (Exception ex) when (!IngestionShutdown.IsInProgress(ex, ct)) { diff --git a/src/Anichron.Worker/Infrastructure/WorkerServiceCollectionExtensions.cs b/src/Anichron.Worker/Infrastructure/WorkerServiceCollectionExtensions.cs index 901f1e6f..1d481415 100644 --- a/src/Anichron.Worker/Infrastructure/WorkerServiceCollectionExtensions.cs +++ b/src/Anichron.Worker/Infrastructure/WorkerServiceCollectionExtensions.cs @@ -66,6 +66,7 @@ internal IServiceCollection AddIngestionServices() services.AddScoped(); services.AddScoped(); services.AddScoped(); + services.AddSingleton(); services.AddScoped(); services.AddScoped(); services.AddScoped(); diff --git a/src/Anichron.Worker/Ingestion/Classification/DuplicateCopyTally.cs b/src/Anichron.Worker/Ingestion/Classification/DuplicateCopyTally.cs new file mode 100644 index 00000000..2cc01f99 --- /dev/null +++ b/src/Anichron.Worker/Ingestion/Classification/DuplicateCopyTally.cs @@ -0,0 +1,12 @@ +using System.Collections.Concurrent; + +namespace Anichron.Worker.Ingestion.Classification; + +internal sealed class DuplicateCopyTally +{ + private readonly ConcurrentDictionary copies = new(); + + public void Count(Guid storageConfigId) => copies.AddOrUpdate(storageConfigId, 1, (_, count) => count + 1); + + public int Take(Guid storageConfigId) => copies.TryRemove(storageConfigId, out var count) ? count : 0; +} diff --git a/src/Anichron.Worker/Ingestion/Classification/KnownFileReconciler.cs b/src/Anichron.Worker/Ingestion/Classification/KnownFileReconciler.cs index 256a4886..88b21836 100644 --- a/src/Anichron.Worker/Ingestion/Classification/KnownFileReconciler.cs +++ b/src/Anichron.Worker/Ingestion/Classification/KnownFileReconciler.cs @@ -3,7 +3,10 @@ namespace Anichron.Worker.Ingestion.Classification; -internal sealed partial class KnownFileReconciler(IClock clock, ILogger logger) +internal sealed partial class KnownFileReconciler( + IClock clock, + DuplicateCopyTally tally, + ILogger logger) { public bool Reconcile(FileClassification file, string relativePath) => file.Outcome switch { @@ -33,6 +36,7 @@ private bool Move(MediaAsset stored, string relativePath) private bool SkipDuplicate(MediaAsset stored, string relativePath) { Log.DuplicateCopy(logger, stored.ContentHash, stored.StorageConfigId, relativePath, stored.FilePath); + tally.Count(stored.StorageConfigId); return false; } @@ -48,7 +52,7 @@ private static partial class Log public static partial void AlreadyIngested(ILogger logger, string relativePath); [LoggerMessage( - Level = LogLevel.Information, + Level = LogLevel.Debug, Message = "Duplicate content {ContentHash} in storage config {StorageConfigId}: {RelativePath} is a copy of {StoredPath}.")] public static partial void DuplicateCopy( ILogger logger, string contentHash, Guid storageConfigId, string relativePath, string storedPath);