diff --git a/CLAUDE.md b/CLAUDE.md index d181b9d3..105a69c8 100644 --- a/CLAUDE.md +++ b/CLAUDE.md @@ -169,6 +169,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`, 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.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..2fb21465 --- /dev/null +++ b/src/Anichron.Worker.Tests.Unit/Crawling/SharedContentReporterTests.cs @@ -0,0 +1,152 @@ +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; + +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 UserStorageConfig Config(Guid id, Guid userId) => new() { Id = id, UserId = userId, RootPath = "/unused" }; + + private sealed class TestFixture + { + 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, Tally, Logger).ReportAsync(crawled, Ct); + + 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_LogsThePairOnceAtDebugAsync() + { + 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")); + + 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 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")); + + 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 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"), + new ContentLocation(Mira, "cafe", Phone, "b.jpg"), + new ContentLocation(Alice, "cafe", Phone, "theirs-too.jpg"), + new ContentLocation(Mira, "beef", Phone, "y.jpg")); + + Pairs(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"), + ]); + } + + // ========================================================================== + // 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_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 9f6559c7..6d1bdee6 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>(crawled => crawled.SequenceEqual(new[] { missing, next })), + 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>(crawled => crawled.SequenceEqual(new[] { home, phone })), + 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 fd7b6cca..6414a344 100644 --- a/src/Anichron.Worker.Tests.Unit/Infrastructure/WorkerServiceCollectionExtensionsTests.cs +++ b/src/Anichron.Worker.Tests.Unit/Infrastructure/WorkerServiceCollectionExtensionsTests.cs @@ -270,6 +270,8 @@ public void TheWholeWorker_ResolvesEveryHostedServiceWithACompleteKindRegistry() { { 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 a59b13c8..09686fef 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,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.Debug && r.Message == "copy/photo.jpg is a copy of already-ingested 2023/photo.jpg, skipping."); + .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] @@ -343,4 +347,23 @@ 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.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 new file mode 100644 index 00000000..66cce671 --- /dev/null +++ b/src/Anichron.Worker/Crawling/SharedContentReporter.cs @@ -0,0 +1,63 @@ +using Anichron.Core.Data.Repository; +using Anichron.Core.Domain; +using Anichron.Worker.Ingestion.Classification; + +namespace Anichron.Worker.Crawling; + +internal interface ISharedContentReporter +{ + Task ReportAsync(IReadOnlyCollection configs, CancellationToken ct); +} + +internal sealed partial class SharedContentReporter( + IMediaAssetRepository repository, + DuplicateCopyTally tally, + ILogger logger) : ISharedContentReporter +{ + public async Task ReportAsync(IReadOnlyCollection configs, CancellationToken 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(); + 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); + 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.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 6ba7fe51..a734394e 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, 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 configs, CancellationToken ct) + { + try + { + await services.GetRequiredService().ReportAsync(configs, 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 0bf94712..e8ced0de 100644 --- a/src/Anichron.Worker/Infrastructure/WorkerServiceCollectionExtensions.cs +++ b/src/Anichron.Worker/Infrastructure/WorkerServiceCollectionExtensions.cs @@ -77,6 +77,7 @@ internal IServiceCollection AddIngestionServices() services.AddSingleton(); services.AddSingleton(); services.AddSingleton(); + services.AddScoped(); services.AddSingleton(); services.AddSingleton(); services.AddImageProxyServices(); @@ -87,6 +88,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 fb910e5c..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 { @@ -32,7 +35,8 @@ 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); + tally.Count(stored.StorageConfigId); return false; } @@ -47,8 +51,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.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); [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);