diff --git a/CLAUDE.md b/CLAUDE.md index 5c67e53..bdc189d 100644 --- a/CLAUDE.md +++ b/CLAUDE.md @@ -132,7 +132,7 @@ do not always match entity names (`AssetInteraction` → `Interactions`, `UserSt - `User` — owns storage configs and interactions - `RefreshToken` — issued and revoked by the API, expired rows deleted by the Worker's `TokenCleanup` work item; `TokenHash` unique - `Invite` — admin-issued registration tokens; `TokenHash` unique, and the only entity carrying an optimistic-concurrency token (`xmin`) -- `WorkItem` — one durable unit of Worker work (`Kind`, jsonb `Payload`, optional `DedupKey`, status, lease, attempt, progress snapshot, result or error); see "Work queue" +- `WorkItem` — one durable unit of Worker work (`Kind`, jsonb `Payload`, optional `DedupKey`, optional `StorageConfigId` scope that cascades on config delete, status, lease, attempt, progress snapshot, result or error); see "Work queue" ### Key EF Core / Database Decisions @@ -159,8 +159,8 @@ The API reads files only through two narrow stores in `Anichron.Infrastructure/S ### Worker Media Processing -- Crawls NAS on a configurable interval (`Worker:CrawlIntervalHours`, default 4 h; immediate crawl on startup) -- The crawl walks folder by folder and never enters `CrawlExclusions`: any dot-folder or dot-file (macOS/Linux/QNAP/ZFS/sync-tool metadata), NAS and OS system folders (`@eaDir`, `#recycle`, `@Recycle`, `$RECYCLE.BIN`, `lost+found`, Nextcloud `files_trashbin`/`files_versions`, …) and app library packages (`*.photoslibrary`, `*.lrdata`, …, which hold thumbnails and renders). `Worker:ExcludedDirectories` adds comma-separated folder names (not paths; rejected at startup) on top. An unreadable subfolder is logged and skipped; an unreadable or missing root fails that storage config's crawl, which is logged as an error, and the Worker moves on to the next config (#282) +- Crawls each storage config as its own `Crawl` work item (#281), due every `Worker:CrawlIntervalHours` (default 4 h) after the last one finished; the first crawl after a start comes from due-ness, so a restart that outlasted the interval crawls straight away. Progress is items found, processed and failed (a Live Photo pair is one item), with no total until the walk ends +- The crawl walks folder by folder and never enters `CrawlExclusions`: any dot-folder or dot-file (macOS/Linux/QNAP/ZFS/sync-tool metadata), NAS and OS system folders (`@eaDir`, `#recycle`, `@Recycle`, `$RECYCLE.BIN`, `lost+found`, Nextcloud `files_trashbin`/`files_versions`, …) and app library packages (`*.photoslibrary`, `*.lrdata`, …, which hold thumbnails and renders). `Worker:ExcludedDirectories` adds comma-separated folder names (not paths; rejected at startup) on top. An unreadable subfolder is logged and skipped; an unreadable or missing root fails that storage config's `Crawl` item with the error, and the other configs' items still run (#282) - Computes XXHash64 `ContentHash` → extracts EXIF via MetadataExtractor → writes `MediaAsset` + `Metadata` - Generates proxy types: `Thumbnail`, `FullPreview`, `WebVideo` (the `ProxyType` enum, persisted as text), and a BlurHash string for images, stored in `Metadata.Blurhash` - Image proxies are auto-oriented once after decoding: upright pixels, orientation tag 1. `Metadata.Orientation` applies only when serving originals — never rotate a proxy by it @@ -169,7 +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 +- Duplicate content is **reported, never merged** (#36). `SharedContentReporter` runs at the end of each config's `Crawl` item: it logs one `Information` summary for the crawled config (copies within it, counted by `DuplicateCopyTally` during that crawl in this Worker, and pairs of its active rows sharing a `ContentHash` with the same user's other active configs), with each pair at `Debug`, and the counts go into the crawl's result. A pair across two configs is reported by each of their crawls. A failed report is logged and does not fail the crawl. 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 @@ -194,12 +194,13 @@ Each user has a private set of `UserStorageConfig` entries pointing to NAS roots ### Work queue -Durable work for the Worker lives in `WorkItems` (#276, ADR 0008). Today its kinds are periodic maintenance (`TokenCleanup`, `WorkItemRetention`); the crawl (#281) and admin requests (#174) follow. +Durable work for the Worker lives in `WorkItems` (#276, ADR 0008). Its kinds are the crawl (`Crawl`, one per storage config) and periodic maintenance (`TokenCleanup`, `WorkItemRetention`); admin requests (#174) follow. - **Producing:** `IWorkQueue.EnqueueAsync` (`EfWorkQueue`, Core) is the only insert. A payload with a `DedupKey` coalesces: while an item with that key is `Queued`, enqueueing returns it (a partial unique index absorbs racing inserts). -- **Scheduling:** every Worker runs `WorkScheduler`. Each poll (`Worker:WorkPollIntervalSeconds`, 5 s) it recovers expired leases, then asks each `IWorkSchedule` whether its kind is due: nothing queued or running with its key and nothing finished within its interval. No timer state is kept, so overdue work runs right after downtime. +- **Scope:** a scoped item carries the `StorageConfigId` it belongs to (`IWorkPayload.StorageConfigId`). A per-user Worker (`WORKER__USER`) claims unscoped items and items of its user's configs; an all-user Worker claims everything. The served user is `WorkerState.ResolvedUserId`, read at each claim and tick, never cached at construction. +- **Scheduling:** every Worker runs `WorkScheduler`. Each poll (`Worker:WorkPollIntervalSeconds`, 5 s) it recovers expired leases, then asks each `IWorkScheduleSource` for its schedules (a fixed `WorkSchedule` is its own source; `CrawlSchedules` returns one per active config the Worker serves) and whether each is due: nothing queued or running with its key and nothing finished within its interval. No timer state is kept, so overdue work runs right after downtime. - **Executing:** `WorkExecutor` claims the oldest queued item (`FOR UPDATE SKIP LOCKED`), one at a time, under a lease (`WorkLeaseSeconds`, 60 s) that a heartbeat (`WorkHeartbeatSeconds`, 2 s) extends while writing the progress snapshot. Every write about a running item is fenced by `(Id, Attempt)`. A fenced heartbeat cancels the handler and writes nothing more; a shutdown releases the item without a stall; a handler exception fails it with no retry. An expired lease requeues the item and counts a stall; the `WorkMaxStalls`th (3) fails it as stalled. An item handed back (release or recovery) while another with its key is already queued ends `Cancelled` as superseded, since the queued one does the same work. A Worker claims only the kinds it knows, so a newer kind waits for a newer Worker. `WorkItemRetention` deletes items finished more than `WorkItemRetentionDays` (30) ago. -- **Adding a kind:** a `WorkKind` value, a payload record in Core implementing `IWorkPayload` (static `Kind`, optional `DedupKey`), an `IWorkHandler` in the Worker registered with `AddWorkHandler`, and for a periodic kind a `WorkSchedule` subclass. `WorkHandlerRegistry` fails startup unless every kind has exactly one payload and one handler; nothing switches on the kind. +- **Adding a kind:** a `WorkKind` value, a payload record in Core implementing `IWorkPayload` (static `Kind`, optional `DedupKey`), an `IWorkHandler` in the Worker registered with `AddWorkHandler`, and for a periodic kind a `WorkSchedule` subclass registered as an `IWorkScheduleSource` (or a source that builds several). `WorkHandlerRegistry` fails startup unless every kind has exactly one payload and one handler; nothing switches on the kind. - **Handler contract:** delivery is at least once. Each run starts from scratch and converges, side effects are safe to repeat, and the returned summary becomes `Result`. ### Worker Processing Model diff --git a/docs/schema.sql b/docs/schema.sql index 8335762..e7af92e 100644 --- a/docs/schema.sql +++ b/docs/schema.sql @@ -64,6 +64,7 @@ CREATE TABLE "WorkItems" ( "Kind" text NOT NULL, "Payload" jsonb NOT NULL, "DedupKey" text, + "StorageConfigId" uuid, "RequestedBy" uuid, "RequestedAt" timestamp with time zone NOT NULL, "Status" text NOT NULL, @@ -80,12 +81,14 @@ CREATE TABLE "WorkItems" ( "Result" jsonb, "Error" text, CONSTRAINT "PK_WorkItems" PRIMARY KEY ("Id"), + CONSTRAINT "FK_WorkItems_StorageConfigs_StorageConfigId" FOREIGN KEY ("StorageConfigId") REFERENCES "StorageConfigs" ("Id") ON DELETE CASCADE, CONSTRAINT "FK_WorkItems_Users_RequestedBy" FOREIGN KEY ("RequestedBy") REFERENCES "Users" ("Id") ON DELETE SET NULL ); COMMENT ON TABLE "WorkItems" IS 'Durable work for the Worker: periodic maintenance today, admin requests later. Only IWorkQueue inserts rows; every later state change belongs to the Worker''s executor. Delivery is at least once, so handlers start from scratch and converge.'; COMMENT ON COLUMN "WorkItems"."Kind" IS 'WorkKind, persisted as text. Picks the payload type and the handler through the registry, never through a switch.'; COMMENT ON COLUMN "WorkItems"."Payload" IS 'The kind''s IWorkPayload, serialised with System.Text.Json and NodaTime settings.'; COMMENT ON COLUMN "WorkItems"."DedupKey" IS 'Opt-in coalescing. While an item with this key is Queued, enqueueing the same key returns that item instead of adding a second; the partial unique index enforces it.'; +COMMENT ON COLUMN "WorkItems"."StorageConfigId" IS 'The storage config the work belongs to, or null for unscoped work. A per-user Worker claims only its user''s configs and unscoped work; deleting the config deletes its items.'; COMMENT ON COLUMN "WorkItems"."RequestedBy" IS 'The admin who asked for the work; null when a Worker''s scheduler enqueued it.'; COMMENT ON COLUMN "WorkItems"."Status" IS 'Queued, Running, Succeeded, Failed or Cancelled, persisted as text.'; COMMENT ON COLUMN "WorkItems"."Attempt" IS 'Incremented by every claim. Every write a Worker makes while running an item is fenced by (Id, Attempt), so an attempt that lost its lease changes nothing.'; @@ -256,6 +259,9 @@ CREATE INDEX ix_workitems_lease ON "WorkItems" ("LeaseUntil") WHERE "Status" = ' CREATE INDEX "IX_WorkItems_RequestedBy" ON "WorkItems" ("RequestedBy"); +CREATE INDEX "IX_WorkItems_StorageConfigId" ON "WorkItems" ("StorageConfigId"); + + CREATE INDEX "IX_WorkItems_UpdatedAt" ON "WorkItems" ("UpdatedAt"); diff --git a/src/Anichron.Core.Tests.Unit/Work/PayloadsTests.cs b/src/Anichron.Core.Tests.Unit/Work/PayloadsTests.cs index 0d1d81f..3baef70 100644 --- a/src/Anichron.Core.Tests.Unit/Work/PayloadsTests.cs +++ b/src/Anichron.Core.Tests.Unit/Work/PayloadsTests.cs @@ -9,6 +9,23 @@ private static (WorkKind Kind, string? DedupKey) Describe(TPayload pay where TPayload : IWorkPayload => (TPayload.Kind, payload.DedupKey); + private static Guid? ScopeOf(TPayload payload) + where TPayload : IWorkPayload + => payload.StorageConfigId; + + [Fact] + public void Crawl_IsOnePeriodicItemPerConfigScopedToIt() + { + var configId = Guid.Parse("77777777-7777-7777-7777-777777777777"); + + Describe(new CrawlPayload(configId)).Should().Be((WorkKind.Crawl, "crawl:77777777-7777-7777-7777-777777777777")); + ScopeOf(new CrawlPayload(configId)).Should().Be(configId); + } + + [Fact] + public void MaintenancePayloads_AreUnscoped() + => (ScopeOf(new TokenCleanupPayload()), ScopeOf(new WorkItemRetentionPayload())).Should().Be((null, null)); + [Fact] public void TokenCleanup_IsOnePeriodicItemAtATime() => Describe(new TokenCleanupPayload()).Should().Be((WorkKind.TokenCleanup, "token-cleanup")); diff --git a/src/Anichron.Core.Tests.Unit/Work/WorkQueueTests.cs b/src/Anichron.Core.Tests.Unit/Work/WorkQueueTests.cs index 75e7a1f..9a21ad8 100644 --- a/src/Anichron.Core.Tests.Unit/Work/WorkQueueTests.cs +++ b/src/Anichron.Core.Tests.Unit/Work/WorkQueueTests.cs @@ -22,6 +22,13 @@ private sealed record Keyed(string Key) : IWorkPayload string IWorkPayload.DedupKey => Key; } + private sealed record Scoped(Guid ConfigId) : IWorkPayload + { + public static WorkKind Kind => WorkKind.TokenCleanup; + + Guid? IWorkPayload.StorageConfigId => ConfigId; + } + private static readonly Instant Now = Instant.FromUtc(2026, 10, 18, 9, 0); private static readonly Guid AdminId = Guid.Parse("11111111-1111-1111-1111-111111111111"); @@ -90,7 +97,18 @@ public async Task EnqueueAsync_Unkeyed_TakesTheKindFromThePayloadTypeAndKeepsNoK await EnqueueAsync(new Unkeyed(1)); var item = (await ItemsAsync()).Should().ContainSingle().Which; - (item.Kind, item.DedupKey, item.RequestedBy).Should().Be((WorkKind.WorkItemRetention, null, null)); + (item.Kind, item.DedupKey, item.RequestedBy, item.StorageConfigId) + .Should().Be((WorkKind.WorkItemRetention, null, null, null)); + } + + [Fact] + public async Task EnqueueAsync_ScopedPayload_StoresItsStorageConfigAsync() + { + var configId = Guid.Parse("55555555-5555-5555-5555-555555555555"); + + await EnqueueAsync(new Scoped(configId)); + + (await ItemsAsync()).Should().ContainSingle().Which.StorageConfigId.Should().Be(configId); } [Fact] diff --git a/src/Anichron.Core/Data/AnichronDbContext.cs b/src/Anichron.Core/Data/AnichronDbContext.cs index c4ec39a..36ed79c 100644 --- a/src/Anichron.Core/Data/AnichronDbContext.cs +++ b/src/Anichron.Core/Data/AnichronDbContext.cs @@ -327,6 +327,9 @@ protected override void OnModelCreating(ModelBuilder modelBuilder) entity.Property(w => w.DedupKey).HasComment( "Opt-in coalescing. While an item with this key is Queued, enqueueing the same key " + "returns that item instead of adding a second; the partial unique index enforces it."); + entity.Property(w => w.StorageConfigId).HasComment( + "The storage config the work belongs to, or null for unscoped work. A per-user Worker " + + "claims only its user's configs and unscoped work; deleting the config deletes its items."); entity.Property(w => w.RequestedBy).HasComment( "The admin who asked for the work; null when a Worker's scheduler enqueued it."); entity.Property(w => w.Status).HasConversion().HasComment( @@ -363,6 +366,11 @@ protected override void OnModelCreating(ModelBuilder modelBuilder) .HasDatabaseName("ix_workitems_lease"); entity.HasIndex(w => w.UpdatedAt); + entity.HasOne() + .WithMany() + .HasForeignKey(w => w.StorageConfigId) + .OnDelete(DeleteBehavior.Cascade); + entity.HasOne() .WithMany() .HasForeignKey(w => w.RequestedBy) diff --git a/src/Anichron.Core/Domain/WorkItem.cs b/src/Anichron.Core/Domain/WorkItem.cs index 66c4c14..0b86852 100644 --- a/src/Anichron.Core/Domain/WorkItem.cs +++ b/src/Anichron.Core/Domain/WorkItem.cs @@ -6,6 +6,7 @@ public class WorkItem public WorkKind Kind { get; set; } public required string Payload { get; set; } public string? DedupKey { get; set; } + public Guid? StorageConfigId { get; set; } public Guid? RequestedBy { get; set; } public Instant RequestedAt { get; set; } diff --git a/src/Anichron.Core/Domain/WorkKind.cs b/src/Anichron.Core/Domain/WorkKind.cs index 3374d5d..17bf019 100644 --- a/src/Anichron.Core/Domain/WorkKind.cs +++ b/src/Anichron.Core/Domain/WorkKind.cs @@ -4,4 +4,5 @@ public enum WorkKind { TokenCleanup = 1, WorkItemRetention = 2, + Crawl = 3, } diff --git a/src/Anichron.Core/Migrations/20261019011459_ScopeWorkItemsToStorageConfigs.Designer.cs b/src/Anichron.Core/Migrations/20261019011459_ScopeWorkItemsToStorageConfigs.Designer.cs new file mode 100644 index 0000000..99c6fa5 --- /dev/null +++ b/src/Anichron.Core/Migrations/20261019011459_ScopeWorkItemsToStorageConfigs.Designer.cs @@ -0,0 +1,692 @@ +// +using System; +using Anichron.Core.Data; +using Microsoft.EntityFrameworkCore; +using Microsoft.EntityFrameworkCore.Infrastructure; +using Microsoft.EntityFrameworkCore.Migrations; +using Microsoft.EntityFrameworkCore.Storage.ValueConversion; +using NodaTime; +using Npgsql.EntityFrameworkCore.PostgreSQL.Metadata; + +#nullable disable + +namespace Anichron.Core.Migrations +{ + [DbContext(typeof(AnichronDbContext))] + [Migration("20261019011459_ScopeWorkItemsToStorageConfigs")] + partial class ScopeWorkItemsToStorageConfigs + { + /// + protected override void BuildTargetModel(ModelBuilder modelBuilder) + { +#pragma warning disable 612, 618 + modelBuilder + .HasAnnotation("ProductVersion", "10.0.13") + .HasAnnotation("Relational:MaxIdentifierLength", 63); + + NpgsqlModelBuilderExtensions.UseIdentityByDefaultColumns(modelBuilder); + + modelBuilder.Entity("Anichron.Core.Domain.AssetInteraction", b => + { + b.Property("Id") + .ValueGeneratedOnAdd() + .HasColumnType("uuid"); + + b.Property("AssetId") + .HasColumnType("uuid"); + + b.Property("IsHidden") + .HasColumnType("boolean"); + + b.Property("IsLiked") + .HasColumnType("boolean"); + + b.Property("IsStarred") + .HasColumnType("boolean"); + + b.Property("LastViewed") + .HasColumnType("timestamp with time zone"); + + b.Property("UserId") + .HasColumnType("uuid"); + + b.Property("ViewCount") + .HasColumnType("integer"); + + b.HasKey("Id"); + + b.HasIndex("AssetId"); + + b.HasIndex("UserId", "AssetId") + .IsUnique(); + + b.ToTable("Interactions", t => + { + t.HasComment("Per-user state for one asset. Never shared between users: two people looking at the same file each get their own row. Drives flashback weighting — hidden excludes, starred always includes and sorts first, liked raises DisplayWeight."); + }); + }); + + modelBuilder.Entity("Anichron.Core.Domain.Burst", b => + { + b.Property("Id") + .ValueGeneratedOnAdd() + .HasColumnType("uuid"); + + b.Property("CreatedAt") + .HasColumnType("timestamp with time zone"); + + b.Property("PrimaryAssetId") + .HasColumnType("uuid") + .HasComment("The frame chosen to represent the burst. Restrict on delete: losing the cover would leave a group that cannot be rendered."); + + b.HasKey("Id"); + + b.HasIndex("PrimaryAssetId"); + + b.ToTable("Bursts", t => + { + t.HasComment("Groups a rapid-fire sequence so a flashback shows one memory rather than nine near-identical frames. The table and schema exist; burst DETECTION is not yet implemented in the Worker, so in a running system this table is empty."); + }); + }); + + modelBuilder.Entity("Anichron.Core.Domain.Invite", b => + { + b.Property("Id") + .ValueGeneratedOnAdd() + .HasColumnType("uuid"); + + b.Property("CreatedAt") + .HasColumnType("timestamp with time zone"); + + b.Property("CreatedByUserId") + .HasColumnType("uuid"); + + b.Property("ExpiresAt") + .HasColumnType("timestamp with time zone"); + + b.Property("TokenHash") + .IsRequired() + .HasMaxLength(44) + .HasColumnType("character varying(44)") + .HasComment("Hash, never the token itself. 44 chars — base64 of a 32-byte value."); + + b.Property("UsedAt") + .HasColumnType("timestamp with time zone"); + + b.Property("UsedByUserId") + .HasColumnType("uuid"); + + b.Property("xmin") + .IsConcurrencyToken() + .ValueGeneratedOnAddOrUpdate() + .HasColumnType("xid") + .HasColumnName("xmin"); + + b.HasKey("Id"); + + b.HasIndex("CreatedByUserId"); + + b.HasIndex("TokenHash") + .IsUnique(); + + b.HasIndex("UsedByUserId"); + + b.ToTable("Invites", t => + { + t.HasComment("Admin-issued registration tokens. Registration requires one; the first user (bootstrap admin) is exempt. Redemption is a read-then-write race, which is why this is the only table carrying a concurrency token."); + }); + }); + + modelBuilder.Entity("Anichron.Core.Domain.MediaAsset", b => + { + b.Property("Id") + .ValueGeneratedOnAdd() + .HasColumnType("uuid"); + + b.Property("BurstId") + .HasColumnType("uuid"); + + b.Property("ContentHash") + .IsRequired() + .HasColumnType("text") + .HasComment("XXHash64 of the file contents. Dedup is deliberately CONFIG-SCOPED: the same bytes under two storage configs are two assets, which is what makes multi-user work. A global unique index on this column alone would break that and must never be added."); + + b.Property("DateCaptured") + .HasColumnType("timestamp without time zone"); + + b.Property("Day") + .HasColumnType("integer") + .HasComment("See Month — the pair backs IX_MediaAsset_Flashback."); + + b.Property("FileName") + .IsRequired() + .HasColumnType("text"); + + b.Property("FilePath") + .IsRequired() + .HasColumnType("text") + .HasComment("Path relative to the storage config root. Not to be confused with ProxyFiles.ProxyPath, which is relative to the proxy root."); + + b.Property("IsSoftDeleted") + .HasColumnType("boolean") + .HasComment("Set by reconciliation when a file is gone from the NAS. Soft rather than hard so the user's interactions survive a temporarily unmounted share."); + + b.Property("LastSeenOnNas") + .HasColumnType("timestamp with time zone"); + + b.Property("MediaType") + .IsRequired() + .HasColumnType("text"); + + b.Property("Month") + .HasColumnType("integer") + .HasComment("Stored apart from DateCaptured so \"On This Day\" is an index lookup on (Month, Day) rather than an EXTRACT() scan over every row."); + + b.Property("PairedAssetId") + .HasColumnType("uuid") + .HasComment("Live Photo pairing: the HEIC parent points at its MOV child. SetNull on delete so losing one half leaves the other a usable standalone asset."); + + b.Property("StorageConfigId") + .HasColumnType("uuid"); + + b.Property("Year") + .HasColumnType("integer"); + + b.HasKey("Id"); + + b.HasIndex("BurstId"); + + b.HasIndex("IsSoftDeleted") + .HasDatabaseName("IX_MediaAssets_Active") + .HasFilter("\"IsSoftDeleted\" = false"); + + b.HasIndex("PairedAssetId"); + + b.HasIndex("Month", "Day") + .HasDatabaseName("IX_MediaAsset_Flashback"); + + b.HasIndex("StorageConfigId", "ContentHash") + .IsUnique() + .HasDatabaseName("ix_mediaassets_storageconfigid_contenthash_active") + .HasFilter("\"IsSoftDeleted\" = false"); + + b.HasIndex("StorageConfigId", "FilePath") + .IsUnique() + .HasDatabaseName("IX_MediaAssets_StorageConfigId_FilePath"); + + b.ToTable("MediaAssets", t => + { + t.HasComment("One row per media file, scoped to a storage config. Written ONLY by the Worker — the API never writes media data."); + }); + }); + + modelBuilder.Entity("Anichron.Core.Domain.Metadata", b => + { + b.Property("AssetId") + .HasColumnType("uuid"); + + b.Property("Blurhash") + .HasColumnType("text") + .HasComment("BlurHash placeholder string (4x3 components), computed by the Worker from the upright image. Kept here rather than as a proxy file so a response needs no disk read for it. Null for videos, and for images ingested before the column existed, which are not backfilled."); + + b.Property("CameraMake") + .HasColumnType("text"); + + b.Property("CameraModel") + .HasColumnType("text"); + + b.Property("DurationInSeconds") + .HasColumnType("integer"); + + b.Property("Height") + .HasColumnType("integer"); + + b.Property("Latitude") + .HasColumnType("double precision"); + + b.Property("LensModel") + .HasColumnType("text"); + + b.Property("Longitude") + .HasColumnType("double precision"); + + b.Property("Orientation") + .ValueGeneratedOnAdd() + .HasColumnType("integer") + .HasDefaultValue(1) + .HasComment("Raw EXIF orientation tag, 1-8 (1 = upright, also used when absent or invalid). Stored raw because 5 and 7 combine a rotation with a mirror and cannot be expressed as degrees; a reader converts, and for 5-8 swaps Width/Height. Proxies keep this tag on unrotated pixels, so do not rotate them twice."); + + b.Property("Width") + .HasColumnType("integer"); + + b.HasKey("AssetId"); + + b.ToTable("Metadata", t => + { + t.HasComment("EXIF data, 1:1 with MediaAsset and keyed by it — there is no separate Id. Populated by ExifExtractionMiddleware; every field is best-effort, because a file may carry no EXIF at all."); + }); + }); + + modelBuilder.Entity("Anichron.Core.Domain.ProxyFile", b => + { + b.Property("Id") + .ValueGeneratedOnAdd() + .HasColumnType("uuid"); + + b.Property("AssetId") + .HasColumnType("uuid"); + + b.Property("CreatedAt") + .HasColumnType("timestamp with time zone"); + + b.Property("ProxyPath") + .IsRequired() + .HasColumnType("text") + .HasComment("Path relative to the proxy root, sharded by CONTENT rather than by ingestion attempt so a retry overwrites its own debris instead of stranding it. IProxyDirectoryStrategy is the single place that rule lives."); + + b.Property("ProxyType") + .IsRequired() + .HasColumnType("text") + .HasComment("Thumbnail, WebVideo or FullPreview, persisted as text. The BlurHash placeholder is not a proxy: it is the Metadata.Blurhash column."); + + b.Property("SizeBytes") + .HasColumnType("bigint"); + + b.HasKey("Id"); + + b.HasIndex("AssetId", "ProxyType") + .IsUnique() + .HasDatabaseName("ix_proxyfiles_assetid_proxytype"); + + b.ToTable("ProxyFiles", t => + { + t.HasComment("One row per generated artefact on the local SSD. An asset has several, but at most one per ProxyType, so of two attempts racing to add the same proxy only one row lands. Written only by ProxyStagingWriter, at the moment the file reaches its final name."); + }); + }); + + modelBuilder.Entity("Anichron.Core.Domain.RefreshToken", b => + { + b.Property("Id") + .ValueGeneratedOnAdd() + .HasColumnType("uuid"); + + b.Property("CreatedAt") + .HasColumnType("timestamp with time zone"); + + b.Property("ExpiresAt") + .HasColumnType("timestamp with time zone"); + + b.Property("RevokedAt") + .HasColumnType("timestamp with time zone"); + + b.Property("TokenHash") + .IsRequired() + .HasColumnType("text") + .HasComment("Hash, never the token itself — a database disclosure must not hand out usable sessions."); + + b.Property("UserId") + .HasColumnType("uuid"); + + b.HasKey("Id"); + + b.HasIndex("TokenHash") + .IsUnique(); + + b.HasIndex("UserId", "ExpiresAt"); + + b.ToTable("RefreshTokens", t => + { + t.HasComment("Issued and revoked by the API; expired rows are swept by the Worker's TokenCleanup work item. One of the few tables both services touch."); + }); + }); + + modelBuilder.Entity("Anichron.Core.Domain.User", b => + { + b.Property("Id") + .ValueGeneratedOnAdd() + .HasColumnType("uuid"); + + b.Property("Email") + .IsRequired() + .HasColumnType("text") + .HasComment("Always stored trimmed and lowercased: an EF value converter applies IdentityNormalizer.Normalize on every write and to every equality comparison against this column, so the case-sensitive unique index cannot hold two spellings of one name."); + + b.Property("FailedLoginAttempts") + .HasColumnType("integer"); + + b.Property("IsAdmin") + .HasColumnType("boolean"); + + b.Property("IsDisabled") + .HasColumnType("boolean"); + + b.Property("LockedUntil") + .HasColumnType("timestamp with time zone"); + + b.Property("MustChangePassword") + .HasColumnType("boolean") + .HasComment("Set by an admin reset. MustChangePasswordMiddleware then 403s every route except the handful needed to change it."); + + b.Property("PasswordHash") + .IsRequired() + .HasColumnType("text") + .HasComment("Argon2id. The parameters live with the hasher, not here, so rotating them does not require a migration."); + + b.Property("Username") + .IsRequired() + .HasColumnType("text") + .HasComment("Always stored trimmed and lowercased: an EF value converter applies IdentityNormalizer.Normalize on every write and to every equality comparison against this column, so the case-sensitive unique index cannot hold two spellings of one name."); + + b.HasKey("Id"); + + b.HasIndex("Email") + .IsUnique() + .HasDatabaseName("ix_users_email"); + + b.HasIndex("Username") + .IsUnique() + .HasDatabaseName("ix_users_username"); + + b.ToTable("Users", t => + { + t.HasComment("Owns storage configs and interactions. All identity writes happen in the API; the Worker only reads."); + }); + }); + + modelBuilder.Entity("Anichron.Core.Domain.UserStorageConfig", b => + { + b.Property("Id") + .ValueGeneratedOnAdd() + .HasColumnType("uuid"); + + b.Property("IsActive") + .HasColumnType("boolean"); + + b.Property("RootPath") + .IsRequired() + .HasColumnType("text") + .HasComment("GLOBALLY unique, not per user: a path cannot be assigned to two configs across all users. Two users indexing the same directory would duplicate every asset and let one user's config removal cascade away the other's rows."); + + b.Property("UserId") + .HasColumnType("uuid"); + + b.HasKey("Id"); + + b.HasIndex("RootPath") + .IsUnique() + .HasDatabaseName("IX_StorageConfigs_RootPath"); + + b.HasIndex("UserId"); + + b.ToTable("StorageConfigs", t => + { + t.HasComment("A NAS root assigned to one user. This is the boundary that scopes assets to a user and lets several Workers each monitor a separate config."); + }); + }); + + modelBuilder.Entity("Anichron.Core.Domain.WorkItem", b => + { + b.Property("Id") + .ValueGeneratedOnAdd() + .HasColumnType("uuid"); + + b.Property("Attempt") + .HasColumnType("integer") + .HasComment("Incremented by every claim. Every write a Worker makes while running an item is fenced by (Id, Attempt), so an attempt that lost its lease changes nothing."); + + b.Property("ClaimedBy") + .HasColumnType("text") + .HasComment("The claiming Worker for display: its WORKER__USER or 'all', then its host name."); + + b.Property("DedupKey") + .HasColumnType("text") + .HasComment("Opt-in coalescing. While an item with this key is Queued, enqueueing the same key returns that item instead of adding a second; the partial unique index enforces it."); + + b.Property("Error") + .HasColumnType("text"); + + b.Property("FinishedAt") + .HasColumnType("timestamp with time zone"); + + b.Property("Kind") + .IsRequired() + .HasColumnType("text") + .HasComment("WorkKind, persisted as text. Picks the payload type and the handler through the registry, never through a switch."); + + b.Property("LeaseUntil") + .HasColumnType("timestamp with time zone") + .HasComment("Extended by every heartbeat. A Running item past it belongs to a dead or paused Worker and goes back to the queue."); + + b.Property("Payload") + .IsRequired() + .HasColumnType("jsonb") + .HasComment("The kind's IWorkPayload, serialised with System.Text.Json and NodaTime settings."); + + b.Property("ProgressDone") + .HasColumnType("bigint"); + + b.Property("ProgressNote") + .HasColumnType("text") + .HasComment("The latest progress snapshot (done, optional total, note), overwritten by each heartbeat; no history is kept."); + + b.Property("ProgressTotal") + .HasColumnType("bigint"); + + b.Property("RequestedAt") + .HasColumnType("timestamp with time zone"); + + b.Property("RequestedBy") + .HasColumnType("uuid") + .HasComment("The admin who asked for the work; null when a Worker's scheduler enqueued it."); + + b.Property("Result") + .HasColumnType("jsonb") + .HasComment("The handler's results summary, written when the item succeeds."); + + b.Property("Stalls") + .HasColumnType("integer") + .HasComment("Lease expiries, counted by lease recovery. A graceful shutdown hands the item back without counting one; the item fails as stalled once the limit is reached."); + + b.Property("StartedAt") + .HasColumnType("timestamp with time zone"); + + b.Property("Status") + .IsRequired() + .HasColumnType("text") + .HasComment("Queued, Running, Succeeded, Failed or Cancelled, persisted as text."); + + b.Property("StorageConfigId") + .HasColumnType("uuid") + .HasComment("The storage config the work belongs to, or null for unscoped work. A per-user Worker claims only its user's configs and unscoped work; deleting the config deletes its items."); + + b.Property("UpdatedAt") + .HasColumnType("timestamp with time zone") + .HasComment("Moves on every insert, claim, heartbeat and transition, so a reader can ask what changed since a point in time."); + + b.HasKey("Id"); + + b.HasIndex("DedupKey") + .IsUnique() + .HasDatabaseName("ix_workitems_dedupkey_queued") + .HasFilter("\"Status\" = 'Queued' AND \"DedupKey\" IS NOT NULL"); + + b.HasIndex("LeaseUntil") + .HasDatabaseName("ix_workitems_lease") + .HasFilter("\"Status\" = 'Running'"); + + b.HasIndex("RequestedAt") + .HasDatabaseName("ix_workitems_claim") + .HasFilter("\"Status\" = 'Queued'"); + + b.HasIndex("RequestedBy"); + + b.HasIndex("StorageConfigId"); + + b.HasIndex("UpdatedAt"); + + b.ToTable("WorkItems", t => + { + t.HasComment("Durable work for the Worker: periodic maintenance today, admin requests later. Only IWorkQueue inserts rows; every later state change belongs to the Worker's executor. Delivery is at least once, so handlers start from scratch and converge."); + }); + }); + + modelBuilder.Entity("Anichron.Core.Domain.AssetInteraction", b => + { + b.HasOne("Anichron.Core.Domain.MediaAsset", "Asset") + .WithMany("Interactions") + .HasForeignKey("AssetId") + .OnDelete(DeleteBehavior.Cascade) + .IsRequired(); + + b.HasOne("Anichron.Core.Domain.User", "User") + .WithMany("Interactions") + .HasForeignKey("UserId") + .OnDelete(DeleteBehavior.Cascade) + .IsRequired(); + + b.Navigation("Asset"); + + b.Navigation("User"); + }); + + modelBuilder.Entity("Anichron.Core.Domain.Burst", b => + { + b.HasOne("Anichron.Core.Domain.MediaAsset", "PrimaryAsset") + .WithMany() + .HasForeignKey("PrimaryAssetId") + .OnDelete(DeleteBehavior.Restrict) + .IsRequired(); + + b.Navigation("PrimaryAsset"); + }); + + modelBuilder.Entity("Anichron.Core.Domain.Invite", b => + { + b.HasOne("Anichron.Core.Domain.User", "CreatedBy") + .WithMany() + .HasForeignKey("CreatedByUserId") + .OnDelete(DeleteBehavior.SetNull); + + b.HasOne("Anichron.Core.Domain.User", "UsedBy") + .WithMany() + .HasForeignKey("UsedByUserId") + .OnDelete(DeleteBehavior.SetNull); + + b.Navigation("CreatedBy"); + + b.Navigation("UsedBy"); + }); + + modelBuilder.Entity("Anichron.Core.Domain.MediaAsset", b => + { + b.HasOne("Anichron.Core.Domain.Burst", "Burst") + .WithMany("Assets") + .HasForeignKey("BurstId") + .OnDelete(DeleteBehavior.SetNull); + + b.HasOne("Anichron.Core.Domain.MediaAsset", "PairedAsset") + .WithMany() + .HasForeignKey("PairedAssetId") + .OnDelete(DeleteBehavior.SetNull); + + b.HasOne("Anichron.Core.Domain.UserStorageConfig", "StorageConfig") + .WithMany("Assets") + .HasForeignKey("StorageConfigId") + .OnDelete(DeleteBehavior.Cascade) + .IsRequired(); + + b.Navigation("Burst"); + + b.Navigation("PairedAsset"); + + b.Navigation("StorageConfig"); + }); + + modelBuilder.Entity("Anichron.Core.Domain.Metadata", b => + { + b.HasOne("Anichron.Core.Domain.MediaAsset", "Asset") + .WithOne("Metadata") + .HasForeignKey("Anichron.Core.Domain.Metadata", "AssetId") + .OnDelete(DeleteBehavior.Cascade) + .IsRequired(); + + b.Navigation("Asset"); + }); + + modelBuilder.Entity("Anichron.Core.Domain.ProxyFile", b => + { + b.HasOne("Anichron.Core.Domain.MediaAsset", "Asset") + .WithMany("ProxyFiles") + .HasForeignKey("AssetId") + .OnDelete(DeleteBehavior.Cascade) + .IsRequired(); + + b.Navigation("Asset"); + }); + + modelBuilder.Entity("Anichron.Core.Domain.RefreshToken", b => + { + b.HasOne("Anichron.Core.Domain.User", "User") + .WithMany("RefreshTokens") + .HasForeignKey("UserId") + .OnDelete(DeleteBehavior.Cascade) + .IsRequired(); + + b.Navigation("User"); + }); + + modelBuilder.Entity("Anichron.Core.Domain.UserStorageConfig", b => + { + b.HasOne("Anichron.Core.Domain.User", "User") + .WithMany("StorageConfigs") + .HasForeignKey("UserId") + .OnDelete(DeleteBehavior.Cascade) + .IsRequired(); + + b.Navigation("User"); + }); + + modelBuilder.Entity("Anichron.Core.Domain.WorkItem", b => + { + b.HasOne("Anichron.Core.Domain.User", null) + .WithMany() + .HasForeignKey("RequestedBy") + .OnDelete(DeleteBehavior.SetNull); + + b.HasOne("Anichron.Core.Domain.UserStorageConfig", null) + .WithMany() + .HasForeignKey("StorageConfigId") + .OnDelete(DeleteBehavior.Cascade); + }); + + modelBuilder.Entity("Anichron.Core.Domain.Burst", b => + { + b.Navigation("Assets"); + }); + + modelBuilder.Entity("Anichron.Core.Domain.MediaAsset", b => + { + b.Navigation("Interactions"); + + b.Navigation("Metadata"); + + b.Navigation("ProxyFiles"); + }); + + modelBuilder.Entity("Anichron.Core.Domain.User", b => + { + b.Navigation("Interactions"); + + b.Navigation("RefreshTokens"); + + b.Navigation("StorageConfigs"); + }); + + modelBuilder.Entity("Anichron.Core.Domain.UserStorageConfig", b => + { + b.Navigation("Assets"); + }); +#pragma warning restore 612, 618 + } + } +} diff --git a/src/Anichron.Core/Migrations/20261019011459_ScopeWorkItemsToStorageConfigs.cs b/src/Anichron.Core/Migrations/20261019011459_ScopeWorkItemsToStorageConfigs.cs new file mode 100644 index 0000000..95b3fd1 --- /dev/null +++ b/src/Anichron.Core/Migrations/20261019011459_ScopeWorkItemsToStorageConfigs.cs @@ -0,0 +1,51 @@ +using System; +using Microsoft.EntityFrameworkCore.Migrations; + +#nullable disable + +namespace Anichron.Core.Migrations +{ + /// + public partial class ScopeWorkItemsToStorageConfigs : Migration + { + /// + protected override void Up(MigrationBuilder migrationBuilder) + { + migrationBuilder.AddColumn( + name: "StorageConfigId", + table: "WorkItems", + type: "uuid", + nullable: true, + comment: "The storage config the work belongs to, or null for unscoped work. A per-user Worker claims only its user's configs and unscoped work; deleting the config deletes its items."); + + migrationBuilder.CreateIndex( + name: "IX_WorkItems_StorageConfigId", + table: "WorkItems", + column: "StorageConfigId"); + + migrationBuilder.AddForeignKey( + name: "FK_WorkItems_StorageConfigs_StorageConfigId", + table: "WorkItems", + column: "StorageConfigId", + principalTable: "StorageConfigs", + principalColumn: "Id", + onDelete: ReferentialAction.Cascade); + } + + /// + protected override void Down(MigrationBuilder migrationBuilder) + { + migrationBuilder.DropForeignKey( + name: "FK_WorkItems_StorageConfigs_StorageConfigId", + table: "WorkItems"); + + migrationBuilder.DropIndex( + name: "IX_WorkItems_StorageConfigId", + table: "WorkItems"); + + migrationBuilder.DropColumn( + name: "StorageConfigId", + table: "WorkItems"); + } + } +} diff --git a/src/Anichron.Core/Migrations/AnichronDbContextModelSnapshot.cs b/src/Anichron.Core/Migrations/AnichronDbContextModelSnapshot.cs index 3d6df1e..347294a 100644 --- a/src/Anichron.Core/Migrations/AnichronDbContextModelSnapshot.cs +++ b/src/Anichron.Core/Migrations/AnichronDbContextModelSnapshot.cs @@ -492,6 +492,10 @@ protected override void BuildModel(ModelBuilder modelBuilder) .HasColumnType("text") .HasComment("Queued, Running, Succeeded, Failed or Cancelled, persisted as text."); + b.Property("StorageConfigId") + .HasColumnType("uuid") + .HasComment("The storage config the work belongs to, or null for unscoped work. A per-user Worker claims only its user's configs and unscoped work; deleting the config deletes its items."); + b.Property("UpdatedAt") .HasColumnType("timestamp with time zone") .HasComment("Moves on every insert, claim, heartbeat and transition, so a reader can ask what changed since a point in time."); @@ -513,6 +517,8 @@ protected override void BuildModel(ModelBuilder modelBuilder) b.HasIndex("RequestedBy"); + b.HasIndex("StorageConfigId"); + b.HasIndex("UpdatedAt"); b.ToTable("WorkItems", t => @@ -643,6 +649,11 @@ protected override void BuildModel(ModelBuilder modelBuilder) .WithMany() .HasForeignKey("RequestedBy") .OnDelete(DeleteBehavior.SetNull); + + b.HasOne("Anichron.Core.Domain.UserStorageConfig", null) + .WithMany() + .HasForeignKey("StorageConfigId") + .OnDelete(DeleteBehavior.Cascade); }); modelBuilder.Entity("Anichron.Core.Domain.Burst", b => diff --git a/src/Anichron.Core/Work/IWorkPayload.cs b/src/Anichron.Core/Work/IWorkPayload.cs index 5f33d80..7497cbc 100644 --- a/src/Anichron.Core/Work/IWorkPayload.cs +++ b/src/Anichron.Core/Work/IWorkPayload.cs @@ -11,4 +11,10 @@ public interface IWorkPayload /// Null, the default, keeps every request a separate item. /// string? DedupKey => null; + + /// + /// The storage config the work is scoped to, or null for work any Worker may run. A per-user + /// Worker claims only its user's configs and unscoped work. + /// + Guid? StorageConfigId => null; } diff --git a/src/Anichron.Core/Work/Payloads.cs b/src/Anichron.Core/Work/Payloads.cs index a619a06..b9977c0 100644 --- a/src/Anichron.Core/Work/Payloads.cs +++ b/src/Anichron.Core/Work/Payloads.cs @@ -15,3 +15,12 @@ public sealed record WorkItemRetentionPayload : IWorkPayload string IWorkPayload.DedupKey => "work-item-retention"; } + +public sealed record CrawlPayload(Guid StorageConfigId) : IWorkPayload +{ + public static WorkKind Kind => WorkKind.Crawl; + + string IWorkPayload.DedupKey => $"crawl:{StorageConfigId}"; + + Guid? IWorkPayload.StorageConfigId => StorageConfigId; +} diff --git a/src/Anichron.Core/Work/WorkQueue.cs b/src/Anichron.Core/Work/WorkQueue.cs index 2b8c378..d9f7116 100644 --- a/src/Anichron.Core/Work/WorkQueue.cs +++ b/src/Anichron.Core/Work/WorkQueue.cs @@ -44,6 +44,7 @@ public async Task EnqueueAsync(TPayload payload, Guid? requested Kind = TPayload.Kind, Payload = WorkJson.Serialize(payload), DedupKey = dedupKey, + StorageConfigId = payload.StorageConfigId, RequestedBy = requestedBy, RequestedAt = now, Status = WorkItemStatus.Queued, diff --git a/src/Anichron.Tests.Integration/Work/MaintenanceHandlerTests.cs b/src/Anichron.Tests.Integration/Work/MaintenanceHandlerTests.cs index f046cb4..3e43a36 100644 --- a/src/Anichron.Tests.Integration/Work/MaintenanceHandlerTests.cs +++ b/src/Anichron.Tests.Integration/Work/MaintenanceHandlerTests.cs @@ -44,7 +44,7 @@ public async Task WorkItemRetention_SecondRun_ConvergesAsync() var work = await WorkDatabase.CreateAsync(server, Ct); var old = WorkDatabase.Start - Duration.FromDays(31); await work.EnqueueAsync(new UnkeyedPayload(1), Ct); - var claimed = (await work.Store.ClaimAsync("worker", old, Lease, Ct))!; + var claimed = (await work.Store.ClaimAsync("worker", null, old, Lease, Ct))!; await work.Store.SucceedAsync(claimed.Attempt, WorkProgress.None, null, old, Ct); var pending = await work.EnqueueAsync(new UnkeyedPayload(2), Ct); diff --git a/src/Anichron.Tests.Integration/Work/WorkItemStoreTests.cs b/src/Anichron.Tests.Integration/Work/WorkItemStoreTests.cs index 5f3dad4..f97f92c 100644 --- a/src/Anichron.Tests.Integration/Work/WorkItemStoreTests.cs +++ b/src/Anichron.Tests.Integration/Work/WorkItemStoreTests.cs @@ -25,7 +25,7 @@ public async Task ClaimAsync_ConcurrentClaimers_NeverGetTheSameItemAsync() await Task.WhenAll(Enumerable.Range(0, 8).Select(_ => Task.Run(async () => { - while (await work.Store.ClaimAsync("worker", At(1), Lease, Ct) is { } item) + while (await work.Store.ClaimAsync("worker", null, At(1), Lease, Ct) is { } item) claimed.Add(item.Attempt.Id); }, Ct))); @@ -43,7 +43,7 @@ public async Task ClaimAsync_OldestItemLockedByAnotherClaim_TakesTheNextWithoutW await using var transaction = await other.Database.BeginTransactionAsync(Ct); await other.Database.ExecuteSqlAsync($"""SELECT 1 FROM "WorkItems" WHERE "Id" = {locked} FOR UPDATE""", Ct); - var claimed = await work.Store.ClaimAsync("worker", At(10), Lease, Ct); + var claimed = await work.Store.ClaimAsync("worker", null, At(10), Lease, Ct); claimed!.Attempt.Id.Should().Be(next); } @@ -56,7 +56,7 @@ public async Task ClaimAsync_TakesTheOldestQueuedItemAndLeasesItAsync() work.Clock.Now = At(1); await work.EnqueueAsync(new UnkeyedPayload(2), Ct); - var claimed = await work.Store.ClaimAsync("all@host", At(10), Lease, Ct); + var claimed = await work.Store.ClaimAsync("all@host", null, At(10), Lease, Ct); claimed.Should().Be(new ClaimedWork(new WorkAttempt(oldest, 1), WorkKind.TokenCleanup, /*lang=json,strict*/ """{"number": 1}""")); var item = await work.ItemAsync(oldest, Ct); @@ -72,7 +72,7 @@ public async Task ClaimAsync_NothingQueued_ReturnsNullAsync() { var work = await WorkDatabase.CreateAsync(server, Ct); - (await work.Store.ClaimAsync("worker", At(0), Lease, Ct)).Should().BeNull(); + (await work.Store.ClaimAsync("worker", null, At(0), Lease, Ct)).Should().BeNull(); } [Fact] @@ -80,7 +80,7 @@ public async Task HeartbeatAsync_HoldingAttempt_ExtendsTheLeaseAndStoresTheProgr { var work = await WorkDatabase.CreateAsync(server, Ct); await work.EnqueueAsync(new UnkeyedPayload(1), Ct); - var claimed = (await work.Store.ClaimAsync("worker", At(0), Lease, Ct))!; + var claimed = (await work.Store.ClaimAsync("worker", null, At(0), Lease, Ct))!; var held = await work.Store.HeartbeatAsync(claimed.Attempt, Halfway, At(30), Lease, Ct); @@ -96,9 +96,9 @@ public async Task AfterAReclaim_TheOldAttemptsHeartbeatAndCompletionChangeNothin { var work = await WorkDatabase.CreateAsync(server, Ct); await work.EnqueueAsync(new UnkeyedPayload(1), Ct); - var zombie = (await work.Store.ClaimAsync("paused", At(0), Lease, Ct))!; + var zombie = (await work.Store.ClaimAsync("paused", null, At(0), Lease, Ct))!; await work.Store.RecoverExpiredLeasesAsync(At(61), 3, Ct); - var current = (await work.Store.ClaimAsync("alive", At(62), Lease, Ct))!; + var current = (await work.Store.ClaimAsync("alive", null, At(62), Lease, Ct))!; var held = await work.Store.HeartbeatAsync(zombie.Attempt, Halfway, At(63), Lease, Ct); await work.Store.SucceedAsync(zombie.Attempt, Halfway, /*lang=json,strict*/ """{"done": true}""", At(64), Ct); @@ -124,7 +124,7 @@ public async Task RecoverExpiredLeasesAsync_RequeuesTwiceThenFailsTheThirdStallA for (var round = 0; round < 3; round++) { var claimedAt = At(round * 100); - await work.Store.ClaimAsync("worker", claimedAt, Lease, Ct); + await work.Store.ClaimAsync("worker", null, claimedAt, Lease, Ct); await work.Store.RecoverExpiredLeasesAsync(claimedAt + Duration.FromSeconds(61), 3, Ct); var item = await work.ItemAsync(id, Ct); stalls.Add((item.Status, item.Stalls)); @@ -141,7 +141,7 @@ public async Task RecoverExpiredLeasesAsync_LeaseStillValid_LeavesTheItemRunning { var work = await WorkDatabase.CreateAsync(server, Ct); var id = await work.EnqueueAsync(new UnkeyedPayload(1), Ct); - await work.Store.ClaimAsync("worker", At(0), Lease, Ct); + await work.Store.ClaimAsync("worker", null, At(0), Lease, Ct); await work.Store.RecoverExpiredLeasesAsync(At(60), 3, Ct); @@ -154,7 +154,7 @@ public async Task RecoverExpiredLeasesAsync_Requeued_ClearsTheClaimAsync() { var work = await WorkDatabase.CreateAsync(server, Ct); var id = await work.EnqueueAsync(new UnkeyedPayload(1), Ct); - await work.Store.ClaimAsync("worker", At(0), Lease, Ct); + await work.Store.ClaimAsync("worker", null, At(0), Lease, Ct); await work.Store.RecoverExpiredLeasesAsync(At(61), 3, Ct); @@ -168,14 +168,14 @@ public async Task ReleaseAsync_QueuesTheItemWithoutAStallAsync() { var work = await WorkDatabase.CreateAsync(server, Ct); await work.EnqueueAsync(new UnkeyedPayload(1), Ct); - var claimed = (await work.Store.ClaimAsync("worker", At(0), Lease, Ct))!; + var claimed = (await work.Store.ClaimAsync("worker", null, At(0), Lease, Ct))!; await work.Store.ReleaseAsync(claimed.Attempt, At(5), Ct); var item = await work.ItemAsync(claimed.Attempt.Id, Ct); (item.Status, item.Stalls, item.Attempt).Should().Be((WorkItemStatus.Queued, 0, 1)); (item.ClaimedBy, item.LeaseUntil).Should().Be((null, null)); - (await work.Store.ClaimAsync("worker", At(6), Lease, Ct))!.Attempt.Should().Be(claimed.Attempt with { Attempt = 2 }); + (await work.Store.ClaimAsync("worker", null, At(6), Lease, Ct))!.Attempt.Should().Be(claimed.Attempt with { Attempt = 2 }); } [Fact] @@ -183,7 +183,7 @@ public async Task ReleaseAsync_SameKeyAlreadyQueued_SupersedesTheReleasedItemAsy { var work = await WorkDatabase.CreateAsync(server, Ct); await work.EnqueueAsync(new KeyedPayload("sweep"), Ct); - var running = (await work.Store.ClaimAsync("worker", At(0), Lease, Ct))!; + var running = (await work.Store.ClaimAsync("worker", null, At(0), Lease, Ct))!; var waiting = await work.EnqueueAsync(new KeyedPayload("sweep"), Ct); await work.Store.ReleaseAsync(running.Attempt, At(5), Ct); @@ -200,10 +200,10 @@ public async Task RecoverExpiredLeasesAsync_SameKeyAlreadyQueued_SupersedesItAnd { var work = await WorkDatabase.CreateAsync(server, Ct); await work.EnqueueAsync(new KeyedPayload("sweep"), Ct); - var keyed = (await work.Store.ClaimAsync("worker", At(0), Lease, Ct))!; + var keyed = (await work.Store.ClaimAsync("worker", null, At(0), Lease, Ct))!; work.Clock.Now = At(1); var other = await work.EnqueueAsync(new UnkeyedPayload(1), Ct); - await work.Store.ClaimAsync("worker", At(1), Lease, Ct); + await work.Store.ClaimAsync("worker", null, At(1), Lease, Ct); var waiting = await work.EnqueueAsync(new KeyedPayload("sweep"), Ct); await work.Store.RecoverExpiredLeasesAsync(At(62), 3, Ct); @@ -220,10 +220,10 @@ public async Task RecoverExpiredLeasesAsync_TwoExpiredItemsShareAKey_RequeuesOne { var work = await WorkDatabase.CreateAsync(server, Ct); await work.EnqueueAsync(new KeyedPayload("sweep"), Ct); - await work.Store.ClaimAsync("worker", At(0), Lease, Ct); + await work.Store.ClaimAsync("worker", null, At(0), Lease, Ct); work.Clock.Now = At(1); await work.EnqueueAsync(new KeyedPayload("sweep"), Ct); - await work.Store.ClaimAsync("worker", At(1), Lease, Ct); + await work.Store.ClaimAsync("worker", null, At(1), Lease, Ct); await work.Store.RecoverExpiredLeasesAsync(At(62), 3, Ct); @@ -248,8 +248,8 @@ await db.Database.ExecuteSqlAsync($""" var known = await work.EnqueueAsync(new UnkeyedPayload(1), Ct); - (await work.Store.ClaimAsync("worker", At(0), Lease, Ct))!.Attempt.Id.Should().Be(known); - (await work.Store.ClaimAsync("worker", At(0), Lease, Ct)).Should().BeNull(); + (await work.Store.ClaimAsync("worker", null, At(0), Lease, Ct))!.Attempt.Id.Should().Be(known); + (await work.Store.ClaimAsync("worker", null, At(0), Lease, Ct)).Should().BeNull(); await using var check = work.CreateDbContext(); (await check.Database .SqlQuery($"""SELECT "Status" AS "Value" FROM "WorkItems" WHERE "Id" = {unknown}""") @@ -261,7 +261,7 @@ public async Task SucceedAsync_StoresTheResultAndTheFinalProgressAsync() { var work = await WorkDatabase.CreateAsync(server, Ct); await work.EnqueueAsync(new UnkeyedPayload(1), Ct); - var claimed = (await work.Store.ClaimAsync("worker", At(0), Lease, Ct))!; + var claimed = (await work.Store.ClaimAsync("worker", null, At(0), Lease, Ct))!; await work.Store.SucceedAsync(claimed.Attempt, Halfway, /*lang=json,strict*/ """{"deleted": 3}""", At(9), Ct); @@ -277,7 +277,7 @@ public async Task FailAsync_StoresTheErrorAndTheLastProgressButNoResultAsync() { var work = await WorkDatabase.CreateAsync(server, Ct); await work.EnqueueAsync(new UnkeyedPayload(1), Ct); - var claimed = (await work.Store.ClaimAsync("worker", At(0), Lease, Ct))!; + var claimed = (await work.Store.ClaimAsync("worker", null, At(0), Lease, Ct))!; await work.Store.FailAsync(claimed.Attempt, Halfway, "InvalidOperationException: boom", At(9), Ct); @@ -312,7 +312,7 @@ public async Task IsDueAsync_FollowsTheLastFinishedItemWithTheKeyAsync() var id = await work.EnqueueAsync(new KeyedPayload("sweep"), Ct); (await work.Store.IsDueAsync("sweep", At(0), Ct)).Should().BeFalse(); - var claimed = (await work.Store.ClaimAsync("worker", At(0), Lease, Ct))!; + var claimed = (await work.Store.ClaimAsync("worker", null, At(0), Lease, Ct))!; (await work.Store.IsDueAsync("sweep", At(0), Ct)).Should().BeFalse(); await work.Store.SucceedAsync(claimed.Attempt, WorkProgress.None, null, At(50), Ct); @@ -324,7 +324,7 @@ public async Task IsDueAsync_FollowsTheLastFinishedItemWithTheKeyAsync() private static async Task FinishedAsync(WorkDatabase work, Instant finishedAt) { await work.EnqueueAsync(new UnkeyedPayload(0), Ct); - var claimed = (await work.Store.ClaimAsync("worker", finishedAt, Lease, Ct))!; + var claimed = (await work.Store.ClaimAsync("worker", null, finishedAt, Lease, Ct))!; await work.Store.SucceedAsync(claimed.Attempt, WorkProgress.None, null, finishedAt, Ct); return claimed.Attempt.Id; } diff --git a/src/Anichron.Tests.Integration/Work/WorkQueueTests.cs b/src/Anichron.Tests.Integration/Work/WorkQueueTests.cs index ed74b5a..a4a6a43 100644 --- a/src/Anichron.Tests.Integration/Work/WorkQueueTests.cs +++ b/src/Anichron.Tests.Integration/Work/WorkQueueTests.cs @@ -25,7 +25,7 @@ public async Task EnqueueAsync_KeyOnlyRunning_AddsANewQueuedItemAsync() { var work = await WorkDatabase.CreateAsync(server, Ct); var running = await work.EnqueueAsync(new KeyedPayload("sweep"), Ct); - await work.Store.ClaimAsync("worker", work.Clock.Now, Lease, Ct); + await work.Store.ClaimAsync("worker", null, work.Clock.Now, Lease, Ct); var queued = await work.EnqueueAsync(new KeyedPayload("sweep"), Ct); diff --git a/src/Anichron.Tests.Integration/Work/WorkScopeTests.cs b/src/Anichron.Tests.Integration/Work/WorkScopeTests.cs new file mode 100644 index 0000000..9bb2d01 --- /dev/null +++ b/src/Anichron.Tests.Integration/Work/WorkScopeTests.cs @@ -0,0 +1,109 @@ +using Anichron.Core.Domain; +using Anichron.Core.Work; +using Microsoft.EntityFrameworkCore; + +namespace Anichron.Tests.Integration.Work; + +public sealed record ScopedPayload(Guid ConfigId) : IWorkPayload +{ + public static WorkKind Kind => WorkKind.TokenCleanup; + + Guid? IWorkPayload.StorageConfigId => ConfigId; +} + +public sealed class WorkScopeTests(PostgresServer server) +{ + private static readonly Duration Lease = Duration.FromSeconds(60); + + private static CancellationToken Ct => TestContext.Current.CancellationToken; + + private static Instant At(int seconds) => WorkDatabase.Start + Duration.FromSeconds(seconds); + + private sealed record Library(WorkDatabase Work, Guid Mira, Guid MiraConfig, Guid Alice, Guid AliceConfig); + + private async Task LibraryAsync() + { + var work = await WorkDatabase.CreateAsync(server, Ct); + var (mira, miraConfig, alice, aliceConfig) = (Guid.NewGuid(), Guid.NewGuid(), Guid.NewGuid(), Guid.NewGuid()); + await using var db = work.CreateDbContext(); + db.Users.AddRange( + new User { Id = mira, Username = "mira", Email = "mira@example.com", PasswordHash = "hash" }, + new User { Id = alice, Username = "alice", Email = "alice@example.com", PasswordHash = "hash" }); + db.StorageConfigs.AddRange( + new UserStorageConfig { Id = miraConfig, UserId = mira, RootPath = "/data/originals/mira", IsActive = true }, + new UserStorageConfig { Id = aliceConfig, UserId = alice, RootPath = "/data/originals/alice", IsActive = true }); + await db.SaveChangesAsync(Ct); + return new Library(work, mira, miraConfig, alice, aliceConfig); + } + + private static async Task EnqueueAtAsync(WorkDatabase work, int seconds, IWorkPayload payload) + { + work.Clock.Now = At(seconds); + return payload switch + { + ScopedPayload scoped => await work.EnqueueAsync(scoped, Ct), + _ => await work.EnqueueAsync((UnkeyedPayload)payload, Ct), + }; + } + + private static async Task> ClaimAllAsync(WorkDatabase work, Guid? servedUserId) + { + var claimed = new List(); + while (await work.Store.ClaimAsync("worker", servedUserId, At(100), Lease, Ct) is { } item) + claimed.Add(item.Attempt.Id); + return claimed; + } + + [Fact] + public async Task EnqueueAsync_StoresThePayloadsScopeAsync() + { + var library = await LibraryAsync(); + + var id = await library.Work.EnqueueAsync(new ScopedPayload(library.MiraConfig), Ct); + + (await library.Work.ItemAsync(id, Ct)).StorageConfigId.Should().Be(library.MiraConfig); + } + + [Fact] + public async Task ClaimAsync_PerUserWorker_TakesItsOwnConfigsAndUnscopedItemsOnlyAsync() + { + var library = await LibraryAsync(); + var alices = await EnqueueAtAsync(library.Work, 0, new ScopedPayload(library.AliceConfig)); + var miras = await EnqueueAtAsync(library.Work, 1, new ScopedPayload(library.MiraConfig)); + var unscoped = await EnqueueAtAsync(library.Work, 2, new UnkeyedPayload(1)); + + var claimed = await ClaimAllAsync(library.Work, library.Mira); + + claimed.Should().Equal(miras, unscoped); + (await library.Work.ItemAsync(alices, Ct)).Status.Should().Be(WorkItemStatus.Queued); + } + + [Fact] + public async Task ClaimAsync_AllUserWorker_TakesEverythingAsync() + { + var library = await LibraryAsync(); + var alices = await EnqueueAtAsync(library.Work, 0, new ScopedPayload(library.AliceConfig)); + var miras = await EnqueueAtAsync(library.Work, 1, new ScopedPayload(library.MiraConfig)); + var unscoped = await EnqueueAtAsync(library.Work, 2, new UnkeyedPayload(1)); + + var claimed = await ClaimAllAsync(library.Work, null); + + claimed.Should().Equal(alices, miras, unscoped); + } + + [Fact] + public async Task DeletingAConfig_DeletesItsItemsAndFencesTheRunningOneAsync() + { + var library = await LibraryAsync(); + await EnqueueAtAsync(library.Work, 0, new ScopedPayload(library.MiraConfig)); + var running = (await library.Work.Store.ClaimAsync("worker", library.Mira, At(1), Lease, Ct))!; + await EnqueueAtAsync(library.Work, 2, new ScopedPayload(library.MiraConfig)); + var alices = await EnqueueAtAsync(library.Work, 3, new ScopedPayload(library.AliceConfig)); + + await using (var db = library.Work.CreateDbContext()) + await db.StorageConfigs.Where(c => c.Id == library.MiraConfig).ExecuteDeleteAsync(Ct); + + (await library.Work.ItemsAsync(Ct)).Select(w => w.Id).Should().Equal(alices); + (await library.Work.Store.HeartbeatAsync(running.Attempt, WorkProgress.None, At(5), Lease, Ct)).Should().BeFalse(); + } +} diff --git a/src/Anichron.Worker.Tests.Unit/Crawling/CrawlTests.cs b/src/Anichron.Worker.Tests.Unit/Crawling/CrawlTests.cs new file mode 100644 index 0000000..9a0c7c3 --- /dev/null +++ b/src/Anichron.Worker.Tests.Unit/Crawling/CrawlTests.cs @@ -0,0 +1,197 @@ +using Anichron.Core.Data.Repository; +using Anichron.Core.Domain; +using Anichron.Core.Work; +using Anichron.Worker.Crawling; +using Anichron.Worker.Settings; +using Anichron.Worker.Work; +using Microsoft.Extensions.DependencyInjection; +using Microsoft.Extensions.Logging; +using Microsoft.Extensions.Logging.Testing; +using Microsoft.Extensions.Options; +using NodaTime; + +namespace Anichron.Worker.Tests.Unit.Crawling; + +public sealed class CrawlTests +{ + private static readonly Guid MiraId = Guid.Parse("66666666-6666-6666-6666-666666666666"); + + private readonly IUserStorageConfigRepository configs = Substitute.For(); + private readonly IFileIngestionPipeline pipeline = Substitute.For(); + private readonly IWorkProgress progress = Substitute.For(); + private readonly ISharedContentReporter reporter = Substitute.For(); + private readonly FakeLogger logger = new(); + + private static CancellationToken Ct => TestContext.Current.CancellationToken; + + private static UserStorageConfig Config(bool active = true) => new() + { + Id = Guid.NewGuid(), + UserId = MiraId, + RootPath = "/data/originals/mira", + IsActive = active, + }; + + private Task RunAsync(Guid configId) + => new CrawlHandler(configs, pipeline, reporter, logger).RunAsync(new CrawlPayload(configId), progress, Ct); + + // ========================================================================== + // CrawlHandler + // ========================================================================== + + [Fact] + public async Task RunAsync_ActiveConfig_CrawlsItAndReturnsTheCountsWithItsDuplicateReportAsync() + { + var config = Config(); + var other = Config(); + configs.FindByIdAsync(config.Id, Ct).Returns(config); + configs.GetActiveByUserIdAsync(MiraId, Ct).Returns([config, other]); + pipeline.RunAsync(config, progress, Ct).Returns(new CrawlResult(5, 4, 1)); + reporter.ReportAsync(config, Arg.Is>(c => c.SequenceEqual(new[] { config, other })), Ct) + .Returns(new DuplicateCounts(2, 1)); + + var result = await RunAsync(config.Id); + + result.Should().Be(new CrawlResult(5, 4, 1) { Duplicates = new DuplicateCounts(2, 1) }); + logger.Collector.GetSnapshot().Should().ContainSingle(e => + e.Level == LogLevel.Information && e.Message.Contains("/data/originals/mira")); + } + + [Fact] + public async Task RunAsync_DuplicateReportFails_LogsItAndStillSucceedsWithoutTheReportAsync() + { + var config = Config(); + configs.FindByIdAsync(config.Id, Ct).Returns(config); + pipeline.RunAsync(config, progress, Ct).Returns(new CrawlResult(1, 1, 0)); + reporter.ReportAsync(Arg.Any(), Arg.Any>(), Arg.Any()) + .Returns(_ => throw new InvalidOperationException("db down")); + + var result = await RunAsync(config.Id); + + result.Should().Be(new CrawlResult(1, 1, 0) { Duplicates = null }); + logger.Collector.GetSnapshot().Should().ContainSingle(e => e.Level == LogLevel.Error) + .Which.Should().Match(e => + e.Message == $"Reporting duplicate content for storage config {config.Id} failed." + && e.Exception is InvalidOperationException); + } + + [Fact] + public async Task RunAsync_ShutdownDuringTheDuplicateReport_PropagatesAsync() + { + var config = Config(); + using var shutdown = new CancellationTokenSource(); + configs.FindByIdAsync(config.Id, shutdown.Token).Returns(config); + pipeline.RunAsync(config, progress, shutdown.Token).Returns(new CrawlResult(1, 1, 0)); + reporter.ReportAsync(Arg.Any(), Arg.Any>(), Arg.Any()) + .Returns(_ => + { + shutdown.Cancel(); + throw new OperationCanceledException(shutdown.Token); + }); + + var act = () => new CrawlHandler(configs, pipeline, reporter, logger) + .RunAsync(new CrawlPayload(config.Id), progress, shutdown.Token); + + await act.Should().ThrowAsync(); + logger.Collector.GetSnapshot().Should().NotContain(e => e.Level == LogLevel.Error); + } + + [Fact] + public async Task RunAsync_InactiveConfig_SkipsWithoutCrawlingAsync() + { + var config = Config(active: false); + configs.FindByIdAsync(config.Id, Ct).Returns(config); + + var result = await RunAsync(config.Id); + + result.Should().Be(new CrawlSkipped("The storage config is missing or inactive.")); + await pipeline.DidNotReceiveWithAnyArgs().RunAsync(default!, default!, Ct); + logger.Collector.GetSnapshot().Should().ContainSingle(e => e.Message.Contains(config.Id.ToString())); + } + + [Fact] + public async Task RunAsync_ConfigGone_SkipsWithoutCrawlingAsync() + { + var result = await RunAsync(Guid.NewGuid()); + + result.Should().BeOfType(); + await pipeline.DidNotReceiveWithAnyArgs().RunAsync(default!, default!, Ct); + } + + [Fact] + public async Task RunAsync_RootMissing_FailsOnlyThisItemByThrowingAsync() + { + var config = Config(); + configs.FindByIdAsync(config.Id, Ct).Returns(config); + pipeline.RunAsync(config, progress, Ct) + .Returns(_ => throw new DirectoryNotFoundException("/data/originals/mira")); + + var act = () => RunAsync(config.Id); + + await act.Should().ThrowAsync(); + await reporter.DidNotReceiveWithAnyArgs().ReportAsync(default!, default!, Ct); + } + + [Fact] + public async Task RunAsync_ShutdownMidCrawl_PropagatesTheCancellationAsync() + { + var config = Config(); + configs.FindByIdAsync(config.Id, Ct).Returns(config); + pipeline.RunAsync(config, progress, Ct).Returns(_ => throw new OperationCanceledException()); + + var act = () => RunAsync(config.Id); + + await act.Should().ThrowAsync(); + } + + // ========================================================================== + // CrawlSchedules + // ========================================================================== + + private static async Task> SchedulesAsync( + CrawlSchedules source, IUserStorageConfigRepository repository) + { + await using var services = new ServiceCollection().AddSingleton(repository).BuildServiceProvider(); + return await source.SchedulesAsync(services, Ct); + } + + [Fact] + public async Task SchedulesAsync_AllUserWorker_SchedulesEveryActiveConfigAtTheCrawlIntervalAsync() + { + var (a, b) = (Config(), Config()); + configs.GetAllActiveAsync(Ct).Returns([a, b]); + var source = new CrawlSchedules(new WorkerState(), Options.Create(new WorkerSettings { CrawlIntervalHours = 6 })); + + var schedules = await SchedulesAsync(source, configs); + + schedules.Select(s => (s.DedupKey, s.Interval)).Should().Equal( + ($"crawl:{a.Id}", Duration.FromHours(6)), ($"crawl:{b.Id}", Duration.FromHours(6))); + } + + [Fact] + public async Task SchedulesAsync_PerUserWorker_SchedulesOnlyItsUsersActiveConfigsAsync() + { + var mine = Config(); + configs.GetActiveByUserIdAsync(MiraId, Ct).Returns([mine]); + configs.GetAllActiveAsync(Ct).Returns([mine, Config()]); + var state = new WorkerState(); + var source = new CrawlSchedules(state, Options.Create(new WorkerSettings())); + state.ResolvedUserId = MiraId; + + var schedules = await SchedulesAsync(source, configs); + + schedules.Should().ContainSingle().Which.Should().BeEquivalentTo( + new { DedupKey = $"crawl:{mine.Id}", Interval = Duration.FromHours(4) }); + } + + [Fact] + public async Task CrawlSchedule_EnqueuesACrawlOfItsConfigAsync() + { + var configId = Guid.NewGuid(); + var queue = Substitute.For(); + + await new CrawlSchedule(configId, Duration.FromHours(4)).EnqueueAsync(queue, Ct); + + await queue.Received(1).EnqueueAsync(new CrawlPayload(configId), null, Ct); + } +} diff --git a/src/Anichron.Worker.Tests.Unit/Crawling/FileIngestionPipelineTests.cs b/src/Anichron.Worker.Tests.Unit/Crawling/FileIngestionPipelineTests.cs index 6380a05..acef7c6 100644 --- a/src/Anichron.Worker.Tests.Unit/Crawling/FileIngestionPipelineTests.cs +++ b/src/Anichron.Worker.Tests.Unit/Crawling/FileIngestionPipelineTests.cs @@ -1,5 +1,6 @@ using Anichron.Core; using Anichron.Core.Domain; +using Anichron.Core.Work; using Anichron.Testing.DependencyInjection; using Anichron.Worker.Crawling; using Anichron.Worker.Ingestion; @@ -16,6 +17,8 @@ namespace Anichron.Worker.Tests.Unit.Crawling; public sealed class FileIngestionPipelineTests { + private static IWorkProgress NoProgress => Substitute.For(); + private static UserStorageConfig MakeConfig(string rootPath = "/nas") => new() { Id = Guid.NewGuid(), @@ -120,7 +123,7 @@ public async Task RunAsync_NasSystemFoldersAndDotFiles_OnlyRealPhotosAreProcesse fixture.FileSystem.AddFile("/nas/.hidden/x.jpg", new MockFileData([])); fixture.FileSystem.AddFile("/nas/._IMG_0001.jpg", new MockFileData([])); - await fixture.Build().RunAsync(MakeConfig(), CancellationToken.None); + await fixture.Build().RunAsync(MakeConfig(), NoProgress, CancellationToken.None); ProcessedRelativePaths(fixture).Should().Equal("2024/IMG_0002.jpg", "IMG_0001.jpg"); } @@ -132,7 +135,7 @@ public async Task RunAsync_ConfiguredExcludedDirectory_IsSkippedAsync() fixture.FileSystem.AddFile("/nas/IMG_0001.jpg", new MockFileData([])); fixture.FileSystem.AddFile("/nas/Lightroom Previews/preview.jpg", new MockFileData([])); - await fixture.Build().RunAsync(MakeConfig(), CancellationToken.None); + await fixture.Build().RunAsync(MakeConfig(), NoProgress, CancellationToken.None); ProcessedRelativePaths(fixture).Should().Equal("IMG_0001.jpg"); } @@ -146,7 +149,7 @@ public async Task RunAsync_UnreadableSubfolder_IsSkippedAndTheCrawlContinuesAsyn fixture.FileSystem.AddFile("/nas/zz-after/IMG_0003.jpg", new MockFileData([])); fixture.FileSystem.AddFile("/nas/aa-before/IMG_0004.jpg", new MockFileData([])); - await fixture.Build().RunAsync(MakeConfig(), CancellationToken.None); + await fixture.Build().RunAsync(MakeConfig(), NoProgress, CancellationToken.None); ProcessedRelativePaths(fixture).Should().Equal("IMG_0001.jpg", "aa-before/IMG_0004.jpg", "zz-after/IMG_0003.jpg"); } @@ -157,7 +160,7 @@ public async Task RunAsync_RootWhoseOwnNameIsExcluded_IsStillCrawledAsync() var fixture = new TestFixture(); fixture.FileSystem.AddFile("/nas/Photos Library.photoslibrary/IMG_0001.jpg", new MockFileData([])); - await fixture.Build().RunAsync(MakeConfig("/nas/Photos Library.photoslibrary"), CancellationToken.None); + await fixture.Build().RunAsync(MakeConfig("/nas/Photos Library.photoslibrary"), NoProgress, CancellationToken.None); ProcessedRelativePaths(fixture).Should().Equal("IMG_0001.jpg"); } @@ -170,7 +173,7 @@ public async Task RunAsync_LivePhotoInNestedFolderNextToEaDir_IsStillPairedAsync fixture.FileSystem.AddFile("/nas/2024/06/IMG_0001.MOV", new MockFileData([])); fixture.FileSystem.AddFile("/nas/2024/06/@eaDir/IMG_0001.HEIC/SYNOPHOTO_THUMB_XL.jpg", new MockFileData([])); - await fixture.Build().RunAsync(MakeConfig(), CancellationToken.None); + await fixture.Build().RunAsync(MakeConfig(), NoProgress, CancellationToken.None); fixture.ProcessedContexts.Should().ContainSingle() .Which.Item.Should().BeOfType(); @@ -188,7 +191,7 @@ public async Task RunAsync_HeicWithMatchingMovSibling_EnqueuesLivePhotoPairItemA fixture.FileSystem.AddFile("/nas/IMG_0001.MOV", new MockFileData([])); var pipeline = fixture.Build(); - await pipeline.RunAsync(MakeConfig("/nas"), CancellationToken.None); + await pipeline.RunAsync(MakeConfig("/nas"), NoProgress, CancellationToken.None); fixture.ProcessedContexts.Should().ContainSingle() .Which.Item.Should().BeOfType(); @@ -202,7 +205,7 @@ public async Task RunAsync_HeicWithMatchingMovSibling_PairItemContainsCorrectPat fixture.FileSystem.AddFile("/nas/IMG_0001.MOV", new MockFileData([])); var pipeline = fixture.Build(); - await pipeline.RunAsync(MakeConfig("/nas"), CancellationToken.None); + await pipeline.RunAsync(MakeConfig("/nas"), NoProgress, CancellationToken.None); var pair = fixture.ProcessedContexts.Single().Item.Should() .BeOfType().Subject; @@ -217,7 +220,7 @@ public async Task RunAsync_HeicWithoutMovSibling_EnqueuesSingleFileItemAsync() fixture.FileSystem.AddFile("/nas/photo.HEIC", new MockFileData([])); var pipeline = fixture.Build(); - await pipeline.RunAsync(MakeConfig("/nas"), CancellationToken.None); + await pipeline.RunAsync(MakeConfig("/nas"), NoProgress, CancellationToken.None); fixture.ProcessedContexts.Should().ContainSingle() .Which.Item.Should().BeOfType(); @@ -231,7 +234,7 @@ public async Task RunAsync_PairedMovFile_IsNotEnqueuedAsSeparateSingleFileItemAs fixture.FileSystem.AddFile("/nas/IMG_0001.MOV", new MockFileData([])); var pipeline = fixture.Build(); - await pipeline.RunAsync(MakeConfig("/nas"), CancellationToken.None); + await pipeline.RunAsync(MakeConfig("/nas"), NoProgress, CancellationToken.None); fixture.ProcessedContexts.Should().ContainSingle(); } @@ -243,7 +246,7 @@ public async Task RunAsync_MovWithoutHeicSibling_EnqueuesSingleFileItemAsync() fixture.FileSystem.AddFile("/nas/video.MOV", new MockFileData([])); var pipeline = fixture.Build(); - await pipeline.RunAsync(MakeConfig("/nas"), CancellationToken.None); + await pipeline.RunAsync(MakeConfig("/nas"), NoProgress, CancellationToken.None); fixture.ProcessedContexts.Should().ContainSingle() .Which.Item.Should().BeOfType() @@ -261,7 +264,7 @@ public async Task RunAsync_UnsupportedFileExtension_IsSkippedAsync() fixture.FileSystem.AddFile("/nas/document.pdf", new MockFileData([])); var pipeline = fixture.Build(); - await pipeline.RunAsync(MakeConfig("/nas"), CancellationToken.None); + await pipeline.RunAsync(MakeConfig("/nas"), NoProgress, CancellationToken.None); fixture.ProcessedContexts.Should().BeEmpty(); } @@ -274,7 +277,7 @@ public async Task RunAsync_MixedSupportedAndUnsupported_OnlySupportedFilesArePro fixture.FileSystem.AddFile("/nas/document.pdf", new MockFileData([])); var pipeline = fixture.Build(); - await pipeline.RunAsync(MakeConfig("/nas"), CancellationToken.None); + await pipeline.RunAsync(MakeConfig("/nas"), NoProgress, CancellationToken.None); fixture.ProcessedContexts.Should().ContainSingle() .Which.Item.Should().BeOfType() @@ -292,7 +295,7 @@ public async Task RunAsync_SingleFileItem_RelativePathIsRelativeToRootPathAsync( fixture.FileSystem.AddFile("/nas/2024/summer/beach.jpg", new MockFileData([])); var pipeline = fixture.Build(); - await pipeline.RunAsync(MakeConfig("/nas"), CancellationToken.None); + await pipeline.RunAsync(MakeConfig("/nas"), NoProgress, CancellationToken.None); var item = fixture.ProcessedContexts.Single().Item.Should() .BeOfType().Subject; @@ -311,7 +314,7 @@ public async Task RunAsync_SameBaseNameInDifferentDirectories_NotPairedAsync() fixture.FileSystem.AddFile("/nas/b/IMG_0001.MOV", new MockFileData([])); var pipeline = fixture.Build(); - await pipeline.RunAsync(MakeConfig("/nas"), CancellationToken.None); + await pipeline.RunAsync(MakeConfig("/nas"), NoProgress, CancellationToken.None); fixture.ProcessedContexts.Should().HaveCount(2); fixture.ProcessedContexts.Should().AllSatisfy(context => @@ -329,7 +332,7 @@ public async Task RunAsync_SingleConsumer_StillIngestsEveryFileAsync() fixture.FileSystem.AddFile("/nas/a.jpg", new MockFileData("a")); fixture.FileSystem.AddFile("/nas/sub/b.jpg", new MockFileData("b")); - await fixture.Build(maxConcurrentFiles: 1).RunAsync(MakeConfig(), CancellationToken.None); + await fixture.Build(maxConcurrentFiles: 1).RunAsync(MakeConfig(), NoProgress, CancellationToken.None); fixture.ProcessedContexts.Select(c => c.Item.RelativePath) .Should().BeEquivalentTo("a.jpg", Path.Combine("sub", "b.jpg")); @@ -342,7 +345,7 @@ public async Task RunAsync_EmptyDirectory_CompletesWithoutExceptionAsync() fixture.FileSystem.AddDirectory("/nas"); var pipeline = fixture.Build(); - var act = async () => await pipeline.RunAsync(MakeConfig("/nas"), CancellationToken.None); + var act = async () => await pipeline.RunAsync(MakeConfig("/nas"), NoProgress, CancellationToken.None); await act.Should().NotThrowAsync(); fixture.ProcessedContexts.Should().BeEmpty(); @@ -363,7 +366,7 @@ public async Task RunAsync_CancellationRequestedWhileIngesting_StopsWithoutTakin var seen = new List(); var pipeline = fixture.BuildWithMiddlewares([new CancellingMiddleware(cts, seen)], maxConcurrentFiles: 1); - var act = async () => await pipeline.RunAsync(MakeConfig("/nas"), cts.Token); + var act = async () => await pipeline.RunAsync(MakeConfig("/nas"), NoProgress, cts.Token); await act.Should().ThrowAsync(); seen.Should().HaveCount(1); @@ -379,12 +382,76 @@ public async Task RunAsync_ItemFails_StillIngestsTheRemainingFilesAsync() var pipeline = fixture.BuildWithMiddlewares( [new FailingOnFirstItemMiddleware(seen)], maxConcurrentFiles: 1); - var act = async () => await pipeline.RunAsync(MakeConfig("/nas"), CancellationToken.None); + var act = async () => await pipeline.RunAsync(MakeConfig("/nas"), NoProgress, CancellationToken.None); await act.Should().NotThrowAsync(); seen.Should().HaveCount(2); } + // ========================================================================== + // Counts and progress + // ========================================================================== + + [Fact] + public async Task RunAsync_CountsItemsFoundProcessedAndFailed_ALivePhotoPairAsOneAsync() + { + var fixture = new TestFixture(); + fixture.FileSystem.AddFile("/nas/a.jpg", new MockFileData([])); + fixture.FileSystem.AddFile("/nas/b.jpg", new MockFileData([])); + fixture.FileSystem.AddFile("/nas/IMG_0001.HEIC", new MockFileData([])); + fixture.FileSystem.AddFile("/nas/IMG_0001.MOV", new MockFileData([])); + fixture.FileSystem.AddFile("/nas/notes.txt", new MockFileData([])); + + var progress = new RecordingProgress(); + + var result = await fixture.BuildWithMiddlewares([new FailingOnFirstItemMiddleware([])], maxConcurrentFiles: 1) + .RunAsync(MakeConfig(), progress, CancellationToken.None); + + result.Should().Be(new CrawlResult(Found: 3, Processed: 2, Failed: 1)); + progress.Reports[^1].Should().Be(new WorkProgress(3, 3, "3 found, 2 processed, 1 failed")); + } + + [Fact] + public async Task RunAsync_ReportsProgressWithoutATotalUntilTheWalkEndsAsync() + { + var fixture = new TestFixture(); + fixture.FileSystem.AddFile("/nas/a.jpg", new MockFileData([])); + fixture.FileSystem.AddFile("/nas/sub/b.jpg", new MockFileData([])); + var progress = new RecordingProgress(); + + await fixture.Build(maxConcurrentFiles: 1).RunAsync(MakeConfig(), progress, CancellationToken.None); + + progress.Reports[0].Should().Be(new WorkProgress(0, null, "1 found, 0 processed, 0 failed")); + progress.Reports[^1].Should().Be(new WorkProgress(2, 2, "2 found, 2 processed, 0 failed")); + progress.Reports.Should().OnlyContain(report => report.Done <= 2); + } + + [Fact] + public async Task RunAsync_SecondRunOverTheSameFolder_ConvergesOnTheSameCountsAsync() + { + var fixture = new TestFixture(); + fixture.FileSystem.AddFile("/nas/a.jpg", new MockFileData([])); + fixture.FileSystem.AddFile("/nas/sub/b.jpg", new MockFileData([])); + var pipeline = fixture.Build(); + var config = MakeConfig(); + + var first = await pipeline.RunAsync(config, NoProgress, CancellationToken.None); + var second = await pipeline.RunAsync(config, NoProgress, CancellationToken.None); + + (first, second).Should().Be((new CrawlResult(2, 2, 0), new CrawlResult(2, 2, 0))); + } + + private sealed class RecordingProgress : IWorkProgress + { + public List Reports { get; } = []; + + public void Report(long done, long? total = null, string? note = null) + { + lock (Reports) + Reports.Add(new WorkProgress(done, total, note)); + } + } + // ========================================================================== // Producer exception // ========================================================================== @@ -395,7 +462,7 @@ public async Task RunAsync_RootDirectoryDoesNotExist_ThrowsDirectoryNotFoundExce var fixture = new TestFixture(); var pipeline = fixture.Build(); - var act = async () => await pipeline.RunAsync(MakeConfig("/nonexistent"), CancellationToken.None); + var act = async () => await pipeline.RunAsync(MakeConfig("/nonexistent"), NoProgress, CancellationToken.None); await act.Should().ThrowAsync(); } diff --git a/src/Anichron.Worker.Tests.Unit/Crawling/SharedContentReporterTests.cs b/src/Anichron.Worker.Tests.Unit/Crawling/SharedContentReporterTests.cs index 2fb2146..3d92671 100644 --- a/src/Anichron.Worker.Tests.Unit/Crawling/SharedContentReporterTests.cs +++ b/src/Anichron.Worker.Tests.Unit/Crawling/SharedContentReporterTests.cs @@ -10,143 +10,162 @@ 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 IReadOnlyCollection? QueriedConfigs { get; private set; } + + private static UserStorageConfig Config(Guid id) => new() { Id = id, UserId = Mira, RootPath = "/unused" }; - public async Task> ReportAsync( - UserStorageConfig[] crawled, params ContentLocation[] locations) + public Task ReportAsync( + Guid crawled, Guid[] usersConfigs, params ContentLocation[] locations) { - Repository.FindSharedContentAsync( - Arg.Is>(ids => ids.SequenceEqual(crawled.Select(c => c.Id))), - Arg.Any()) - .Returns(locations); + Repository.FindSharedContentAsync(Arg.Any>(), Arg.Any()) + .Returns(call => + { + QueriedConfigs = call.Arg>(); + return locations; + }); + + return new SharedContentReporter(Repository, Tally, Logger) + .ReportAsync(Config(crawled), [.. usersConfigs.Select(Config)], Ct); + } - await new SharedContentReporter(Repository, Tally, Logger).ReportAsync(crawled, Ct); + public IEnumerable Pairs => Logger.Collector.GetSnapshot().Where(r => r.Level == LogLevel.Debug); - return Logger.Collector.GetSnapshot(); - } + public IEnumerable Summaries + => Logger.Collector.GetSnapshot().Where(r => r.Level == LogLevel.Information).Select(r => r.Message); } - private static IEnumerable Pairs(IEnumerable records) - => records.Where(r => r.Level == LogLevel.Debug); + [Fact] + public async Task ReportAsync_LooksAcrossTheCrawledConfigAndTheUsersOtherConfigsAsync() + { + var fx = new TestFixture(); - private static IEnumerable Summaries(IEnumerable records) - => records.Where(r => r.Level == LogLevel.Information).Select(r => r.Message); + await fx.ReportAsync(Phone, [Home, Backup]); - // ========================================================================== - // Pairs across configs - // ========================================================================== + fx.QueriedConfigs.Should().Equal(Home, Backup, Phone); + } + + [Fact] + public async Task ReportAsync_CrawledConfigAlreadyAmongTheUsersConfigs_QueriesItOnceAsync() + { + var fx = new TestFixture(); + + await fx.ReportAsync(Home, [Home, Phone]); + + fx.QueriedConfigs.Should().Equal(Home, Phone); + } [Fact] public async Task ReportAsync_ContentInTwoConfigs_LogsThePairOnceAtDebugAsync() { - var records = await new TestFixture().ReportAsync( - [Config(Home, Mira), Config(Phone, Mira)], + var fx = new TestFixture(); + + var counts = await fx.ReportAsync( + Home, [Home, Phone], new ContentLocation(Mira, "cafe", Phone, "dcim/a.jpg"), new ContentLocation(Mira, "cafe", Home, "2023/a.jpg")); - Pairs(records).Should().ContainSingle().Which.Message.Should().Be( + counts.Should().Be(new DuplicateCounts(0, 1)); + fx.Summaries.Should().ContainSingle().Which.Should().EndWith( + "copies within it: 0, pairs with the user's other storage configs: 1."); + fx.Pairs.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() + public async Task ReportAsync_ContentInThreeConfigs_LogsOnlyThePairsInvolvingTheCrawledConfigAsync() { - var records = await new TestFixture().ReportAsync( - [Config(Home, Mira), Config(Phone, Mira), Config(Backup, Mira)], + var fx = new TestFixture(); + + var counts = await fx.ReportAsync( + Phone, [Home, Phone, Backup], 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")); + counts.PairsAcross.Should().Be(2); + fx.Pairs.Select(r => (r.GetStructuredStateValue("FilePath"), r.GetStructuredStateValue("OtherFilePath"))) + .Should().Equal(("a.jpg", "b.jpg"), ("b.jpg", "c.jpg")); } [Fact] - public async Task ReportAsync_TwoHashesAndTwoUsers_PairsOnlyWithinEachUsersHashAsync() + public async Task ReportAsync_TwoHashes_PairsOnlyEqualContentAsync() { - var records = await new TestFixture().ReportAsync( - [Config(Home, Mira), Config(Phone, Mira), Config(Backup, Alice)], + var fx = new TestFixture(); + + await fx.ReportAsync( + Home, [Home, Phone], 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"), - ]); + fx.Pairs.Select(r => (r.GetStructuredStateValue("ContentHash"), r.GetStructuredStateValue("FilePath"))) + .Should().BeEquivalentTo([("cafe", "a.jpg"), ("beef", "x.jpg")]); } - // ========================================================================== - // Summary per user - // ========================================================================== - [Fact] - public async Task ReportAsync_CopiesWithinAndAcrossConfigs_SummarisesEachUserOnceAtInformationAsync() + public async Task ReportAsync_CopiesWithinAndPairsAcross_SummarisesTheCrawlOnceAtInformationAsync() { 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)], + var counts = await fx.ReportAsync( + Home, [Home, Phone], 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."); + counts.Should().Be(new DuplicateCounts(2, 1)); + fx.Summaries.Should().Equal( + "Duplicate content in storage config 11111111-0000-0000-0000-000000000000 of user " + + "aaaaaaaa-0000-0000-0000-000000000001: copies within it: 2, pairs with the user's other storage configs: 1."); } [Fact] - public async Task ReportAsync_OnlyAPairAcrossConfigs_StillSummarisesTheUserAsync() + public async Task ReportAsync_OnlyCopiesWithin_StillSummarisesAsync() { - 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")); + var fx = new TestFixture(); + fx.Tally.Count(Home); + + (await fx.ReportAsync(Home, [Home])).Should().Be(new DuplicateCounts(1, 0)); - Summaries(records).Should().ContainSingle().Which.Should().EndWith( - "copies within one storage config: 0, pairs across storage configs: 1."); + fx.Summaries.Should().ContainSingle().Which.Should().EndWith( + "copies within it: 1, pairs with the user's other storage configs: 0."); } [Fact] - public async Task ReportAsync_Reported_StartsTheNextCycleFromZeroAsync() + public async Task ReportAsync_TakesOnlyTheCrawledConfigsCopiesAsync() { var fx = new TestFixture(); fx.Tally.Count(Home); + fx.Tally.Count(Phone); - await fx.ReportAsync([Config(Home, Mira)]); + await fx.ReportAsync(Home, [Home, Phone]); - fx.Tally.Take(Home).Should().Be(0); + (fx.Tally.Take(Home), fx.Tally.Take(Phone)).Should().Be((0, 1)); } [Fact] public async Task ReportAsync_NothingDuplicated_LogsNothingAsync() - => (await new TestFixture().ReportAsync([Config(Home, Mira), Config(Backup, Alice)])).Should().BeEmpty(); + { + var fx = new TestFixture(); + + (await fx.ReportAsync(Home, [Home, Phone])).Should().Be(new DuplicateCounts(0, 0)); + + fx.Logger.Collector.GetSnapshot().Should().BeEmpty(); + } } diff --git a/src/Anichron.Worker.Tests.Unit/Crawling/WorkerTests.cs b/src/Anichron.Worker.Tests.Unit/Crawling/WorkerTests.cs deleted file mode 100644 index 6d1bdee..0000000 --- a/src/Anichron.Worker.Tests.Unit/Crawling/WorkerTests.cs +++ /dev/null @@ -1,326 +0,0 @@ -using Anichron.Core.Data.Repository; -using Anichron.Core.Domain; -using Anichron.Testing.DependencyInjection; -using Anichron.Testing.Time; -using Anichron.Worker.Crawling; -using Anichron.Worker.Settings; -using Microsoft.Extensions.DependencyInjection; -using Microsoft.Extensions.Logging; -using Microsoft.Extensions.Logging.Testing; -using Microsoft.Extensions.Options; -using Microsoft.Extensions.Time.Testing; -using CrawlingWorker = Anichron.Worker.Crawling.Worker; - -namespace Anichron.Worker.Tests.Unit.Crawling; - -public sealed class WorkerTests -{ - private const int IntervalHours = 3; - - private static CancellationToken Ct => TestContext.Current.CancellationToken; - - 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) - .AddSingleton(Reporter)); - - ConfigRepository.GetAllActiveAsync(Arg.Any()) - .Returns([]); - ConfigRepository.GetActiveByUserIdAsync(Arg.Any(), Arg.Any()) - .Returns([]); - } - - public FakeLogger Logger { get; } = new(); - - public CrawlingWorker Build(TimeProvider? timeProvider = null) - => new( - scopeFactory, - WorkerState, - Pipeline, - timeProvider ?? new FakeTimeProvider(), - Options.Create(new WorkerSettings { CrawlIntervalHours = IntervalHours }), - Logger); - } - - // ========================================================================== - // ExecuteAsync — crawl schedule - // ========================================================================== - - [Fact] - public async Task ExecuteAsync_Started_CrawlsAtOnceThenWaitsTheConfiguredIntervalAsync() - { - await using var timeline = new Timeline(); - var worker = await StartAsync(timeline); - - await timeline.ExpectAsync(new Called("crawl"), new TimerArmed(TimeSpan.FromHours(3))); - worker.ExecuteTask!.IsCompleted.Should().BeFalse(); - } - - [Fact] - public async Task ExecuteAsync_Stopped_EndsWithoutCrawlingAgainAsync() - { - await using var timeline = new Timeline(); - var worker = await StartAsync(timeline); - await timeline.ExpectAsync(new Called("crawl"), new TimerArmed(TimeSpan.FromHours(3))); - - await worker.StopAsync(Ct); - timeline.Clock.Advance(TimeSpan.FromHours(3)); - - await timeline.ExpectAsync(new Finished(TaskStatus.Canceled)); - timeline.Unread.Should().BeEmpty(); - } - - private static async Task StartAsync(Timeline timeline) - { - var fixture = new TestFixture(); - fixture.ConfigRepository.GetAllActiveAsync(Arg.Any()) - .Returns(call => - { - call.Arg().ThrowIfCancellationRequested(); - timeline.Record("crawl"); - return []; - }); - var worker = fixture.Build(timeline.Clock); - await worker.StartAsync(Ct); - timeline.Watch(worker.ExecuteTask!, () => worker.StopAsync(CancellationToken.None)); - return worker; - } - - // ========================================================================== - // All-user mode - // ========================================================================== - - [Fact] - public async Task CrawlAllAsync_NoResolvedUserId_CallsGetAllActiveAsync() - { - var fixture = new TestFixture(); - var crawlingWorker = fixture.Build(); - - await crawlingWorker.CrawlAllAsync(CancellationToken.None); - - await fixture.ConfigRepository.Received(1).GetAllActiveAsync(Arg.Any()); - await fixture.ConfigRepository.DidNotReceive() - .GetActiveByUserIdAsync(Arg.Any(), Arg.Any()); - } - - // ========================================================================== - // Dedicated-user mode - // ========================================================================== - - [Fact] - public async Task CrawlAllAsync_WithResolvedUserId_CallsGetActiveByUserIdAsync() - { - var fixture = new TestFixture(); - var userId = Guid.NewGuid(); - fixture.WorkerState.ResolvedUserId = userId; - var crawlingWorker = fixture.Build(); - - await crawlingWorker.CrawlAllAsync(CancellationToken.None); - - await fixture.ConfigRepository.Received(1) - .GetActiveByUserIdAsync(userId, Arg.Any()); - await fixture.ConfigRepository.DidNotReceive().GetAllActiveAsync(Arg.Any()); - } - - // ========================================================================== - // Pipeline invocation - // ========================================================================== - - [Fact] - public async Task CrawlAllAsync_MultipleActiveConfigs_RunsPipelineForEachAsync() - { - var fixture = new TestFixture(); - var configs = new List - { - new() { Id = Guid.NewGuid(), UserId = Guid.NewGuid(), RootPath = "/nas/a" }, - new() { Id = Guid.NewGuid(), UserId = Guid.NewGuid(), RootPath = "/nas/b" }, - }; - fixture.ConfigRepository.GetAllActiveAsync(Arg.Any()).Returns(configs); - var crawlingWorker = fixture.Build(); - - await crawlingWorker.CrawlAllAsync(CancellationToken.None); - - await fixture.Pipeline.Received(2) - .RunAsync(Arg.Any(), Arg.Any()); - } - - [Fact] - public async Task CrawlAllAsync_NoActiveConfigs_PipelineIsNeverCalledAsync() - { - var fixture = new TestFixture(); - var crawlingWorker = fixture.Build(); - - await crawlingWorker.CrawlAllAsync(CancellationToken.None); - - await fixture.Pipeline.DidNotReceive() - .RunAsync(Arg.Any(), Arg.Any()); - } - - [Fact] - public async Task CrawlAllAsync_ActiveConfig_RunsPipelineWithCorrectConfigAsync() - { - var fixture = new TestFixture(); - var config = new UserStorageConfig { Id = Guid.NewGuid(), UserId = Guid.NewGuid(), RootPath = "/nas/photos" }; - fixture.ConfigRepository.GetAllActiveAsync(Arg.Any()) - .Returns([config]); - var crawlingWorker = fixture.Build(); - - await crawlingWorker.CrawlAllAsync(CancellationToken.None); - - await fixture.Pipeline.Received(1).RunAsync(config, Arg.Any()); - } - - // ========================================================================== - // A failing config - // ========================================================================== - - [Fact] - public async Task CrawlAllAsync_ConfigCrawlFails_LogsItsRootAndCrawlsTheNextConfigAsync() - { - var fixture = new TestFixture(); - var userId = Guid.NewGuid(); - var missing = new UserStorageConfig { Id = Guid.NewGuid(), UserId = userId, RootPath = "/data/originals/mira" }; - var next = new UserStorageConfig { Id = Guid.NewGuid(), UserId = Guid.NewGuid(), RootPath = "/data/originals/alice" }; - fixture.ConfigRepository.GetAllActiveAsync(Arg.Any()).Returns([missing, next]); - fixture.Pipeline.RunAsync(missing, Arg.Any()) - .Returns(Task.FromException(new DirectoryNotFoundException("gone"))); - var crawlingWorker = fixture.Build(); - - 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." - && r.Exception is DirectoryNotFoundException); - } - - [Fact] - public async Task CrawlAllAsync_ShutdownDuringACrawl_StopsWithoutLoggingAFailureAsync() - { - var fixture = new TestFixture(); - var first = new UserStorageConfig { Id = Guid.NewGuid(), UserId = Guid.NewGuid(), RootPath = "/data/originals/mira" }; - var second = new UserStorageConfig { Id = Guid.NewGuid(), UserId = Guid.NewGuid(), RootPath = "/data/originals/alice" }; - fixture.ConfigRepository.GetAllActiveAsync(Arg.Any()).Returns([first, second]); - using var shutdown = new CancellationTokenSource(); - await shutdown.CancelAsync(); - fixture.Pipeline.RunAsync(first, Arg.Any()).Returns(Task.FromCanceled(shutdown.Token)); - var crawlingWorker = fixture.Build(); - - var act = () => crawlingWorker.CrawlAllAsync(shutdown.Token); - - 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); - } - - [Fact] - public async Task CrawlAllAsync_CancellationWithoutShutdown_IsLoggedAsACrawlFailureAsync() - { - var fixture = new TestFixture(); - var config = new UserStorageConfig { Id = Guid.NewGuid(), UserId = Guid.NewGuid(), RootPath = "/data/originals/mira" }; - fixture.ConfigRepository.GetAllActiveAsync(Arg.Any()).Returns([config]); - fixture.Pipeline.RunAsync(config, Arg.Any()) - .Returns(Task.FromException(new OperationCanceledException("timed out"))); - var crawlingWorker = fixture.Build(); - - await crawlingWorker.CrawlAllAsync(Ct); - - fixture.Logger.Collector.GetSnapshot().Should().ContainSingle(r => r.Level == LogLevel.Error) - .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() - { - await using var timeline = new Timeline(); - var fixture = new TestFixture(); - var config = new UserStorageConfig { Id = Guid.NewGuid(), UserId = Guid.NewGuid(), RootPath = "/data/originals/mira" }; - fixture.ConfigRepository.GetAllActiveAsync(Arg.Any()).Returns([config]); - fixture.Pipeline.RunAsync(config, Arg.Any()) - .Returns(_ => - { - timeline.Record("crawl"); - return Task.FromException(new DirectoryNotFoundException("gone")); - }); - var worker = fixture.Build(timeline.Clock); - await worker.StartAsync(Ct); - timeline.Watch(worker.ExecuteTask!, () => worker.StopAsync(CancellationToken.None)); - await timeline.ExpectAsync(new Called("crawl"), new TimerArmed(TimeSpan.FromHours(3))); - - timeline.Clock.Advance(TimeSpan.FromHours(3)); - - await timeline.ExpectAsync( - new TimerFired(TimeSpan.FromHours(3)), new Called("crawl"), new TimerArmed(TimeSpan.FromHours(3))); - worker.ExecuteTask!.IsCompleted.Should().BeFalse(); - } -} diff --git a/src/Anichron.Worker.Tests.Unit/Infrastructure/WorkerServiceCollectionExtensionsTests.cs b/src/Anichron.Worker.Tests.Unit/Infrastructure/WorkerServiceCollectionExtensionsTests.cs index 6414a34..af90e4e 100644 --- a/src/Anichron.Worker.Tests.Unit/Infrastructure/WorkerServiceCollectionExtensionsTests.cs +++ b/src/Anichron.Worker.Tests.Unit/Infrastructure/WorkerServiceCollectionExtensionsTests.cs @@ -199,15 +199,14 @@ public void AddIngestionServices_RegistersProxyStagingWriter_AsSingleton() // ========================================================================== [Fact] - public void AddWorkerHostedServices_MigratesFirstThenRunsTheWorkQueueAndTheCrawl() + public void AddWorkerHostedServices_MigratesFirstThenRunsTheWorkQueue() { var services = new ServiceCollection(); services.AddWorkerHostedServices(); services.Where(d => d.ServiceType == typeof(IHostedService)).Select(d => d.ImplementationType).Should().Equal( - typeof(DatabaseMigratorService), typeof(WorkerInitializer), typeof(WorkScheduler), typeof(WorkExecutor), - typeof(Worker.Crawling.Worker)); + typeof(DatabaseMigratorService), typeof(WorkerInitializer), typeof(WorkScheduler), typeof(WorkExecutor)); } // ========================================================================== @@ -225,16 +224,18 @@ public void AddWorkQueueServices_RegistersEachKindsHandlerAsScoped() .Select(d => (d.ServiceType, d.ImplementationType)).Should().BeEquivalentTo([ (typeof(IWorkHandler), typeof(TokenCleanupHandler)), (typeof(IWorkHandler), typeof(WorkItemRetentionHandler)), + (typeof(IWorkHandler), typeof(CrawlHandler)), ]); } [Fact] public void AddWorkQueueServices_RegistersEachPeriodicKindsSchedule() => new ServiceCollection().AddWorkQueueServices() - .Where(d => d.ServiceType == typeof(IWorkSchedule)) + .Where(d => d.ServiceType == typeof(IWorkScheduleSource)) .Select(d => (d.ImplementationType, d.Lifetime)).Should().BeEquivalentTo([ (typeof(TokenCleanupSchedule), ServiceLifetime.Singleton), (typeof(WorkItemRetentionSchedule), ServiceLifetime.Singleton), + (typeof(CrawlSchedules), ServiceLifetime.Singleton), ]); [Fact] @@ -254,7 +255,7 @@ public void TheWholeWorker_ResolvesEveryHostedServiceWithACompleteKindRegistry() var hosted = provider.GetServices(); - hosted.Should().HaveCount(5); + hosted.Should().HaveCount(4); provider.GetRequiredService().Should().NotBeNull(); using var scope = provider.CreateScope(); provider.GetServices().Select(r => r.Kind).Should().BeEquivalentTo(Enum.GetValues()); diff --git a/src/Anichron.Worker.Tests.Unit/Work/WorkExecutorTests.cs b/src/Anichron.Worker.Tests.Unit/Work/WorkExecutorTests.cs index 0030e09..5570d2d 100644 --- a/src/Anichron.Worker.Tests.Unit/Work/WorkExecutorTests.cs +++ b/src/Anichron.Worker.Tests.Unit/Work/WorkExecutorTests.cs @@ -2,6 +2,7 @@ using Anichron.Core.Work; using Anichron.Testing.DependencyInjection; using Anichron.Testing.Time; +using Anichron.Worker.Crawling; using Anichron.Worker.Settings; using Anichron.Worker.Work; using Microsoft.Extensions.DependencyInjection; @@ -24,6 +25,11 @@ public sealed record Retention : IWorkPayload public static WorkKind Kind => WorkKind.WorkItemRetention; } + public sealed record Crawl : IWorkPayload + { + public static WorkKind Kind => WorkKind.Crawl; + } + private static readonly TimeSpan Heartbeat = TimeSpan.FromSeconds(2); private static readonly TimeSpan Poll = TimeSpan.FromSeconds(5); private static readonly Duration Lease = Duration.FromSeconds(60); @@ -33,6 +39,7 @@ public sealed record Retention : IWorkPayload private readonly IWorkHandler handler = Substitute.For>(); private readonly FakeLogger logger = new(); private readonly Queue queued = new(); + private readonly WorkerState workerState = new(); private readonly List> runs = []; private CancellationToken handlerToken; private IWorkProgress? handlerProgress; @@ -40,7 +47,7 @@ public sealed record Retention : IWorkPayload public WorkExecutorTests() { - store.ClaimAsync(Arg.Any(), Arg.Any(), Arg.Any(), Arg.Any()) + store.ClaimAsync(Arg.Any(), Arg.Any(), Arg.Any(), Arg.Any(), Arg.Any()) .Returns(_ => { timeline.Record("claim"); @@ -88,10 +95,10 @@ private async Task StartAsync(params ClaimedWork[] items) foreach (var item in items) queued.Enqueue(item); var registry = new WorkHandlerRegistry( - [WorkHandlerRegistration.For(), WorkHandlerRegistration.For()], - [typeof(Cleanup), typeof(Retention)]); + [WorkHandlerRegistration.For(), WorkHandlerRegistration.For(), WorkHandlerRegistration.For()], + [typeof(Cleanup), typeof(Retention), typeof(Crawl)]); executor = new WorkExecutor( - store, registry, ScopeFactories.Providing(services => services.AddSingleton(handler)), + store, registry, workerState, ScopeFactories.Providing(services => services.AddSingleton(handler)), timeline.Clock, Options.Create(new WorkerSettings()), logger); await executor.StartAsync(Ct); timeline.Watch(executor.ExecuteTask!, () => executor.StopAsync(CancellationToken.None)); @@ -124,7 +131,24 @@ public async Task ExecuteAsync_ClaimsAsThisWorkerWithTheConfiguredLeaseAsync() await StartAsync(); await timeline.ExpectAsync(new Called("claim"), new TimerArmed(Poll)); - await store.Received(1).ClaimAsync($"all@{Environment.MachineName}", Now, Lease, Arg.Any()); + await store.Received(1).ClaimAsync($"all@{Environment.MachineName}", null, Now, Lease, Arg.Any()); + } + + [Fact] + public async Task ExecuteAsync_ClaimsForTheUserResolvedAtThatMomentAsync() + { + var mira = Guid.Parse("44444444-4444-4444-4444-444444444444"); + await StartAsync(); + await timeline.ExpectAsync(new Called("claim"), new TimerArmed(Poll)); + + workerState.ResolvedUserId = mira; + timeline.Clock.Advance(Poll); + + await timeline.ExpectAsync(new TimerFired(Poll), new Called("claim"), new TimerArmed(Poll)); + store.ReceivedCalls() + .Where(call => call.GetMethodInfo().Name == nameof(IWorkItemStore.ClaimAsync)) + .Select(call => (Guid?)call.GetArguments()[1]) + .Should().Equal(null, mira); } [Theory] @@ -137,7 +161,7 @@ public void ClaimantName_IsTheWorkersUserOrAllThenItsHost(string user, string ex public async Task ExecuteAsync_ClaimFails_LogsAndClaimsAgainAtTheNextPollAsync() { var calls = 0; - store.ClaimAsync(Arg.Any(), Arg.Any(), Arg.Any(), Arg.Any()) + store.ClaimAsync(Arg.Any(), Arg.Any(), Arg.Any(), Arg.Any(), Arg.Any()) .Returns(_ => { timeline.Record("claim"); diff --git a/src/Anichron.Worker.Tests.Unit/Work/WorkHandlerRegistryTests.cs b/src/Anichron.Worker.Tests.Unit/Work/WorkHandlerRegistryTests.cs index eb6b0f9..d1edc1f 100644 --- a/src/Anichron.Worker.Tests.Unit/Work/WorkHandlerRegistryTests.cs +++ b/src/Anichron.Worker.Tests.Unit/Work/WorkHandlerRegistryTests.cs @@ -17,10 +17,15 @@ public sealed record NumberedRetentionPayload(int Number) : IWorkPayload public static WorkKind Kind => WorkKind.WorkItemRetention; } - private static readonly Type[] RealPayloads = [typeof(TokenCleanupPayload), typeof(WorkItemRetentionPayload)]; + private static readonly Type[] RealPayloads = + [typeof(TokenCleanupPayload), typeof(WorkItemRetentionPayload), typeof(CrawlPayload)]; private static WorkHandlerRegistration[] RealHandlers - => [WorkHandlerRegistration.For(), WorkHandlerRegistration.For()]; + => [ + WorkHandlerRegistration.For(), + WorkHandlerRegistration.For(), + WorkHandlerRegistration.For(), + ]; private static Action Building(IEnumerable registrations, params Type[] payloads) => () => _ = new WorkHandlerRegistry(registrations, payloads); @@ -40,12 +45,14 @@ public void Constructor_FromDependencyInjection_ChecksTheCorePayloads() var act = () => new WorkHandlerRegistry([WorkHandlerRegistration.For()]); act.Should().Throw() - .WithMessage("WorkKind.WorkItemRetention needs exactly one handler, found none."); + .Which.Message.Should().Be(string.Join(Environment.NewLine, + "WorkKind.WorkItemRetention needs exactly one handler, found none.", + "WorkKind.Crawl needs exactly one handler, found none.")); } [Fact] public void Constructor_KindWithoutPayload_FailsNamingIt() - => Building(RealHandlers, typeof(TokenCleanupPayload)).Should().Throw() + => Building(RealHandlers, typeof(TokenCleanupPayload), typeof(CrawlPayload)).Should().Throw() .WithMessage("WorkKind.WorkItemRetention needs exactly one payload type, found none."); [Fact] @@ -61,7 +68,7 @@ public void Constructor_KindWithTwoHandlers_FailsNamingThem() [Fact] public void Constructor_HandlerForAnotherPayloadOfTheKind_FailsNamingBoth() - => Building([WorkHandlerRegistration.For(), RealHandlers[1]], RealPayloads) + => Building([WorkHandlerRegistration.For(), RealHandlers[1], RealHandlers[2]], RealPayloads) .Should().Throw() .WithMessage("WorkKind.TokenCleanup's handler takes OtherCleanupPayload, not its payload TokenCleanupPayload."); @@ -70,7 +77,8 @@ public void Constructor_SeveralProblems_ReportsEachOnItsOwnLine() => Building([], RealPayloads).Should().Throw() .Which.Message.Should().Be(string.Join(Environment.NewLine, "WorkKind.TokenCleanup needs exactly one handler, found none.", - "WorkKind.WorkItemRetention needs exactly one handler, found none.")); + "WorkKind.WorkItemRetention needs exactly one handler, found none.", + "WorkKind.Crawl needs exactly one handler, found none.")); [Fact] public async Task RunAsync_DispatchesTheDeserialisedPayloadToTheKindsHandlerAsync() @@ -81,8 +89,8 @@ public async Task RunAsync_DispatchesTheDeserialisedPayloadToTheKindsHandlerAsyn handler.RunAsync(new NumberedRetentionPayload(7), progress, cts.Token).Returns("summary"); await using var services = new ServiceCollection().AddSingleton(handler).BuildServiceProvider(); var registry = new WorkHandlerRegistry( - [RealHandlers[0], WorkHandlerRegistration.For()], - [typeof(TokenCleanupPayload), typeof(NumberedRetentionPayload)]); + [RealHandlers[0], WorkHandlerRegistration.For(), RealHandlers[2]], + [typeof(TokenCleanupPayload), typeof(NumberedRetentionPayload), typeof(CrawlPayload)]); var work = new ClaimedWork(new WorkAttempt(Guid.NewGuid(), 1), WorkKind.WorkItemRetention, /*lang=json,strict*/ """{"number":7}"""); var result = await registry.RunAsync(work, services, progress, cts.Token); diff --git a/src/Anichron.Worker.Tests.Unit/Work/WorkSchedulerTests.cs b/src/Anichron.Worker.Tests.Unit/Work/WorkSchedulerTests.cs index a7617c6..72c8721 100644 --- a/src/Anichron.Worker.Tests.Unit/Work/WorkSchedulerTests.cs +++ b/src/Anichron.Worker.Tests.Unit/Work/WorkSchedulerTests.cs @@ -24,6 +24,12 @@ public sealed record Keyed(string Key) : IWorkPayload private sealed class KeyedSchedule(string key, Duration interval) : WorkSchedule(new Keyed(key), interval); + private sealed class ChangingSource(Func> current) : IWorkScheduleSource + { + public Task> SchedulesAsync(IServiceProvider services, CancellationToken ct) + => Task.FromResult(current()); + } + private static readonly TimeSpan Poll = TimeSpan.FromSeconds(5); private static readonly Duration Hourly = Duration.FromHours(1); private static readonly Duration Daily = Duration.FromDays(1); @@ -59,10 +65,10 @@ public WorkSchedulerTests() private Instant Now => Instant.FromDateTimeOffset(timeline.Clock.GetUtcNow()); - private async Task StartAsync(params IWorkSchedule[] schedules) + private async Task StartAsync(params IWorkScheduleSource[] sources) { scheduler = new WorkScheduler( - store, schedules, ScopeFactories.Providing(services => services.AddSingleton(queue)), + store, sources, ScopeFactories.Providing(services => services.AddSingleton(queue)), timeline.Clock, Options.Create(new WorkerSettings()), logger); await scheduler.StartAsync(Ct); timeline.Watch(scheduler.ExecuteTask!, () => scheduler.StopAsync(CancellationToken.None)); @@ -104,6 +110,30 @@ await timeline.ExpectAsync( await store.Received(1).IsDueAsync("sweep", Now - Hourly, Arg.Any()); } + [Fact] + public async Task ExecuteAsync_AsksEachSourceForItsSchedulesAtEveryTickAsync() + { + IReadOnlyList current = [new KeyedSchedule("crawl:a", Hourly)]; + due.UnionWith(["crawl:a", "crawl:b"]); + await StartAsync(new ChangingSource(() => current)); + await timeline.ExpectAsync(new Called("recover"), new Called("enqueue crawl:a"), new TimerArmed(Poll)); + current = [new KeyedSchedule("crawl:a", Hourly), new KeyedSchedule("crawl:b", Hourly)]; + + timeline.Clock.Advance(Poll); + + await timeline.ExpectAsync( + new TimerFired(Poll), new Called("recover"), new Called("enqueue crawl:a"), new Called("enqueue crawl:b"), + new TimerArmed(Poll)); + } + + [Fact] + public async Task WorkSchedule_IsItsOwnOnlySourceAsync() + { + var schedule = new KeyedSchedule("sweep", Hourly); + + (await schedule.SchedulesAsync(Substitute.For(), Ct)).Should().Equal(schedule); + } + [Fact] public async Task ExecuteAsync_TickFails_LogsAndTicksAgainAtTheNextPollAsync() { diff --git a/src/Anichron.Worker/Crawling/Crawl.cs b/src/Anichron.Worker/Crawling/Crawl.cs new file mode 100644 index 0000000..e4a5bbb --- /dev/null +++ b/src/Anichron.Worker/Crawling/Crawl.cs @@ -0,0 +1,75 @@ +using Anichron.Core.Data.Repository; +using Anichron.Core.Domain; +using Anichron.Core.Work; +using Anichron.Worker.Ingestion; +using Anichron.Worker.Settings; +using Anichron.Worker.Work; +using Microsoft.Extensions.Options; +using NodaTime; + +namespace Anichron.Worker.Crawling; + +public sealed record CrawlSkipped(string Reason); + +internal sealed partial class CrawlHandler( + IUserStorageConfigRepository configs, + IFileIngestionPipeline pipeline, + ISharedContentReporter reporter, + ILogger logger) : IWorkHandler +{ + public async Task RunAsync(CrawlPayload payload, IWorkProgress progress, CancellationToken ct) + { + var config = await configs.FindByIdAsync(payload.StorageConfigId, ct); + if (config is not { IsActive: true }) + { + Log.Skipped(logger, payload.StorageConfigId); + return new CrawlSkipped("The storage config is missing or inactive."); + } + + Log.Crawling(logger, config.RootPath, config.UserId); + var counts = await pipeline.RunAsync(config, progress, ct); + return counts with { Duplicates = await ReportDuplicatesAsync(config, ct) }; + } + + private async Task ReportDuplicatesAsync(UserStorageConfig config, CancellationToken ct) + { + try + { + var usersConfigs = await configs.GetActiveByUserIdAsync(config.UserId, ct); + return await reporter.ReportAsync(config, usersConfigs, ct); + } + catch (Exception ex) when (!IngestionShutdown.IsInProgress(ex, ct)) + { + Log.DuplicateReportFailed(logger, config.Id, ex); + return null; + } + } + + private static partial class Log + { + [LoggerMessage(Level = LogLevel.Information, Message = "Crawling {RootPath} (userId={UserId}).")] + public static partial void Crawling(ILogger logger, string rootPath, Guid userId); + + [LoggerMessage(Level = LogLevel.Error, Message = "Reporting duplicate content for storage config {Id} failed.")] + public static partial void DuplicateReportFailed(ILogger logger, Guid id, Exception ex); + + [LoggerMessage(Level = LogLevel.Information, Message = "Skipping the crawl of storage config {Id}: it is missing or inactive.")] + public static partial void Skipped(ILogger logger, Guid id); + } +} + +public sealed class CrawlSchedule(Guid storageConfigId, Duration interval) + : WorkSchedule(new CrawlPayload(storageConfigId), interval); + +public sealed class CrawlSchedules(WorkerState workerState, IOptions options) : IWorkScheduleSource +{ + public async Task> SchedulesAsync(IServiceProvider services, CancellationToken ct) + { + var repository = services.GetRequiredService(); + var served = workerState.ResolvedUserId is { } userId + ? await repository.GetActiveByUserIdAsync(userId, ct) + : await repository.GetAllActiveAsync(ct); + var interval = Duration.FromHours(options.Value.CrawlIntervalHours); + return [.. served.Select(config => new CrawlSchedule(config.Id, interval))]; + } +} diff --git a/src/Anichron.Worker/Crawling/CrawlTally.cs b/src/Anichron.Worker/Crawling/CrawlTally.cs new file mode 100644 index 0000000..3029e84 --- /dev/null +++ b/src/Anichron.Worker/Crawling/CrawlTally.cs @@ -0,0 +1,51 @@ +using Anichron.Core.Work; + +namespace Anichron.Worker.Crawling; + +/// A crawl's results summary. A Live Photo pair counts as one item. +/// Items the walk queued for ingestion. +/// Items ingested without an error. +/// Items whose ingestion failed. +public sealed record CrawlResult(long Found, long Processed, long Failed) +{ + /// The duplicate report for the crawled config, or null when the report failed. + public DuplicateCounts? Duplicates { get; init; } +} + +internal sealed class CrawlTally(IWorkProgress progress) +{ + private readonly Lock gate = new(); + private long found; + private long processed; + private long failed; + private bool walked; + + public CrawlResult Result + { + get + { + lock (gate) + return new CrawlResult(found, processed, failed); + } + } + + public void Found() => Update(() => found++); + + public void Walked() => Update(() => walked = true); + + public void Processed() => Update(() => processed++); + + public void Failed() => Update(() => failed++); + + private void Update(Action change) + { + lock (gate) + { + change(); + progress.Report( + processed + failed, + walked ? found : null, + $"{found} found, {processed} processed, {failed} failed"); + } + } +} diff --git a/src/Anichron.Worker/Crawling/FileIngestionPipeline.cs b/src/Anichron.Worker/Crawling/FileIngestionPipeline.cs index 584fca5..6a4e1b0 100644 --- a/src/Anichron.Worker/Crawling/FileIngestionPipeline.cs +++ b/src/Anichron.Worker/Crawling/FileIngestionPipeline.cs @@ -1,5 +1,6 @@ using Anichron.Core; using Anichron.Core.Domain; +using Anichron.Core.Work; using Anichron.Worker.Ingestion; using Anichron.Worker.Ingestion.Pipeline; using Anichron.Worker.Settings; @@ -11,7 +12,7 @@ namespace Anichron.Worker.Crawling; internal interface IFileIngestionPipeline { - Task RunAsync(UserStorageConfig config, CancellationToken ct); + Task RunAsync(UserStorageConfig config, IWorkProgress progress, CancellationToken ct); } internal sealed partial class FileIngestionPipeline( @@ -25,8 +26,9 @@ internal sealed partial class FileIngestionPipeline( { private readonly WorkerSettings settings = workerOptions.Value; - public Task RunAsync(UserStorageConfig config, CancellationToken ct) + public async Task RunAsync(UserStorageConfig config, IWorkProgress progress, CancellationToken ct) { + var tally = new CrawlTally(progress); var channel = Channel.CreateBounded( // Stryker disable once Arithmetic,Initializer new BoundedChannelOptions(settings.MaxConcurrentFiles * 2) @@ -36,18 +38,20 @@ public Task RunAsync(UserStorageConfig config, CancellationToken ct) FullMode = BoundedChannelFullMode.Wait, }); - var producer = ProduceAsync(config, channel.Writer, ct); + var producer = ProduceAsync(config, channel.Writer, tally, ct); var consumers = Enumerable .Range(0, settings.MaxConcurrentFiles) - .Select(workerIndex => ConsumeAsync(config, channel.Reader, workerIndex, ct)) + .Select(workerIndex => ConsumeAsync(config, channel.Reader, workerIndex, tally, ct)) .ToArray(); - return Task.WhenAll([producer, .. consumers]); + await Task.WhenAll([producer, .. consumers]); + return tally.Result; } private async Task ProduceAsync( UserStorageConfig config, ChannelWriter writer, + CrawlTally tally, CancellationToken ct) { try @@ -56,10 +60,12 @@ private async Task ProduceAsync( while (pending.TryPop(out var directory)) { var (files, subdirectories) = ListDirectory(directory, config.RootPath); - await EnqueueDirectoryItemsAsync(files, config.RootPath, writer, ct); + await EnqueueDirectoryItemsAsync(files, config.RootPath, writer, tally, ct); foreach (var subdirectory in subdirectories) pending.Push(subdirectory); } + + tally.Walked(); } finally { @@ -88,12 +94,16 @@ private async Task EnqueueDirectoryItemsAsync( IList filesInDirectory, string rootPath, ChannelWriter writer, + CrawlTally tally, CancellationToken ct) { var linkResult = livePhotoLinker.Link(filesInDirectory, rootPath); foreach (var item in linkResult.Items) + { + tally.Found(); await writer.WriteAsync(item, ct); + } foreach (var filePath in filesInDirectory) { @@ -108,6 +118,7 @@ private async Task EnqueueDirectoryItemsAsync( } var relativePath = fileSystem.Path.GetRelativePath(rootPath, filePath); + tally.Found(); await writer.WriteAsync(new SingleFileItem(filePath, relativePath, mediaType.Value), ct); } } @@ -116,6 +127,7 @@ private async Task ConsumeAsync( UserStorageConfig config, ChannelReader reader, int workerIndex, + CrawlTally tally, CancellationToken ct) { await foreach (var item in reader.ReadAllAsync(ct)) @@ -130,6 +142,7 @@ private async Task ConsumeAsync( { var context = new IngestionContext { Item = item, Config = config, AttemptId = guidFactory.NewGuid() }; await runner.RunAsync(context, ct); + tally.Processed(); } catch (Exception ex) when (IngestionShutdown.IsInProgress(ex, ct)) { @@ -139,6 +152,7 @@ private async Task ConsumeAsync( catch (Exception ex) { Log.ItemFailed(logger, item.AbsolutePath, ex); + tally.Failed(); } } } diff --git a/src/Anichron.Worker/Crawling/SharedContentReporter.cs b/src/Anichron.Worker/Crawling/SharedContentReporter.cs index 66cce67..a2814d1 100644 --- a/src/Anichron.Worker/Crawling/SharedContentReporter.cs +++ b/src/Anichron.Worker/Crawling/SharedContentReporter.cs @@ -4,9 +4,15 @@ namespace Anichron.Worker.Crawling; +/// Duplicate content found by one config's crawl. +/// Second copies of stored content at another path in the crawled config. +/// Pairs of the same content in the crawled config and another of the user's configs. +public sealed record DuplicateCounts(int CopiesWithin, int PairsAcross); + internal interface ISharedContentReporter { - Task ReportAsync(IReadOnlyCollection configs, CancellationToken ct); + Task ReportAsync( + UserStorageConfig crawled, IReadOnlyCollection usersConfigs, CancellationToken ct); } internal sealed partial class SharedContentReporter( @@ -14,34 +20,32 @@ internal sealed partial class SharedContentReporter( DuplicateCopyTally tally, ILogger logger) : ISharedContentReporter { - public async Task ReportAsync(IReadOnlyCollection configs, CancellationToken ct) + public async Task ReportAsync( + UserStorageConfig crawled, IReadOnlyCollection usersConfigs, 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 = tally.Take(crawled.Id); + var configIds = usersConfigs.Select(config => config.Id).Union([crawled.Id]).ToList(); + var locations = await repository.FindSharedContentAsync(configIds, ct); + + var pairs = 0; + foreach (var content in locations.GroupBy(location => 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) + var places = content.OrderBy(location => location.StorageConfigId).ToList(); + var crawledPairs = places + .SelectMany((first, index) => places.Skip(index + 1).Select(second => (first, second))) + .Where(pair => pair.first.StorageConfigId == crawled.Id || pair.second.StorageConfigId == crawled.Id); + foreach (var (first, second) in crawledPairs) { Log.SharedContent( logger, first.ContentHash, first.UserId, first.FilePath, first.StorageConfigId, second.FilePath, second.StorageConfigId); - pairsByUser[first.UserId] = pairsByUser.GetValueOrDefault(first.UserId) + 1; + pairs++; } } - foreach (var (userId, copies) in copiesByUser) - { - var pairs = pairsByUser.GetValueOrDefault(userId); - if (copies + pairs > 0) - Log.Summary(logger, userId, copies, pairs); - } + if (copies + pairs > 0) + Log.Summary(logger, crawled.Id, crawled.UserId, copies, pairs); + return new DuplicateCounts(copies, pairs); } private static partial class Log @@ -56,8 +60,9 @@ public static partial void SharedContent( [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); + Message = "Duplicate content in storage config {StorageConfigId} of user {UserId}: copies within it: " + + "{WithinConfigCopies}, pairs with the user's other storage configs: {AcrossConfigPairs}.")] + public static partial void Summary( + ILogger logger, Guid storageConfigId, Guid userId, int withinConfigCopies, int acrossConfigPairs); } } diff --git a/src/Anichron.Worker/Crawling/Worker.cs b/src/Anichron.Worker/Crawling/Worker.cs deleted file mode 100644 index a734394..0000000 --- a/src/Anichron.Worker/Crawling/Worker.cs +++ /dev/null @@ -1,85 +0,0 @@ -using Anichron.Core.Data.Repository; -using Anichron.Core.Domain; -using Anichron.Worker.Ingestion; -using Anichron.Worker.Scheduling; -using Anichron.Worker.Settings; -using Microsoft.Extensions.Options; - -namespace Anichron.Worker.Crawling; - -internal sealed partial class Worker( - IServiceScopeFactory scopeFactory, - WorkerState workerState, - IFileIngestionPipeline ingestionPipeline, - TimeProvider timeProvider, - IOptions options, - ILogger logger) : BackgroundService -{ - private readonly WorkerSettings settings = options.Value; - - protected override Task ExecuteAsync(CancellationToken stoppingToken) - { - Log.Starting(logger, workerState.ResolvedUserId is { } userId - ? $"dedicated (userId={userId})" - : "all-user"); - - return RecurringWork.RunAsync( - CrawlAllAsync, TimeSpan.FromHours(settings.CrawlIntervalHours), timeProvider, stoppingToken); - } - - internal async Task CrawlAllAsync(CancellationToken ct) - { - using var scope = scopeFactory.CreateScope(); - var repository = scope.ServiceProvider.GetRequiredService(); - - var configs = workerState.ResolvedUserId is { } userId - ? await repository.GetActiveByUserIdAsync(userId, ct) - : await repository.GetAllActiveAsync(ct); - - foreach (var config in configs) - await CrawlAsync(config, ct); - - await ReportSharedContentAsync(scope.ServiceProvider, configs, ct); - } - - private async Task CrawlAsync(UserStorageConfig config, CancellationToken ct) - { - Log.CrawlingPath(logger, config.RootPath, config.UserId); - try - { - await ingestionPipeline.RunAsync(config, ct); - } - catch (Exception ex) when (!IngestionShutdown.IsInProgress(ex, ct)) - { - Log.CrawlFailed(logger, config.RootPath, config.UserId, ex); - } - } - - 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.")] - public static partial void Starting(ILogger logger, string mode); - - [LoggerMessage(Level = LogLevel.Information, Message = "Crawling {RootPath} (userId={UserId}).")] - public static partial void CrawlingPath(ILogger logger, string rootPath, Guid userId); - - [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/Crawling/WorkerState.cs b/src/Anichron.Worker/Crawling/WorkerState.cs index 664271a..22d0c22 100644 --- a/src/Anichron.Worker/Crawling/WorkerState.cs +++ b/src/Anichron.Worker/Crawling/WorkerState.cs @@ -2,7 +2,7 @@ namespace Anichron.Worker.Crawling; public sealed class WorkerState { - // Written once by WorkerInitializer during startup; safe to read in Worker.ExecuteAsync - // because IHostedService.StartAsync is called sequentially in registration order. + // Written once by WorkerInitializer during startup. Read it at claim and schedule time, never + // in a constructor: hosted services are built before any of them starts. public Guid? ResolvedUserId { get; set; } } diff --git a/src/Anichron.Worker/Infrastructure/WorkerServiceCollectionExtensions.cs b/src/Anichron.Worker/Infrastructure/WorkerServiceCollectionExtensions.cs index e8ced0d..62cfbcb 100644 --- a/src/Anichron.Worker/Infrastructure/WorkerServiceCollectionExtensions.cs +++ b/src/Anichron.Worker/Infrastructure/WorkerServiceCollectionExtensions.cs @@ -16,7 +16,6 @@ using Microsoft.EntityFrameworkCore; using NodaTime; using System.IO.Abstractions; -using CrawlingWorker = Anichron.Worker.Crawling.Worker; namespace Anichron.Worker.Infrastructure; @@ -58,8 +57,10 @@ internal IServiceCollection AddWorkQueueServices() services.AddSingleton(); services.AddWorkHandler(); services.AddWorkHandler(); - services.AddSingleton(); - services.AddSingleton(); + services.AddWorkHandler(); + services.AddSingleton(); + services.AddSingleton(); + services.AddSingleton(); return services; } @@ -123,7 +124,6 @@ internal IServiceCollection AddWorkerHostedServices() services.AddHostedService(); services.AddHostedService(); services.AddHostedService(); - services.AddHostedService(); return services; } } diff --git a/src/Anichron.Worker/Work/EfWorkItemStore.cs b/src/Anichron.Worker/Work/EfWorkItemStore.cs index 99cc929..a354b63 100644 --- a/src/Anichron.Worker/Work/EfWorkItemStore.cs +++ b/src/Anichron.Worker/Work/EfWorkItemStore.cs @@ -15,7 +15,8 @@ public sealed class EfWorkItemStore(IDbContextFactory context private sealed record ClaimRow(Guid Id, string Kind, string Payload, int Attempt); // Stryker disable all - public async Task ClaimAsync(string claimedBy, Instant now, Duration lease, CancellationToken ct) + public async Task ClaimAsync( + string claimedBy, Guid? servedUserId, Instant now, Duration lease, CancellationToken ct) { await using var db = await contexts.CreateDbContextAsync(ct); var leaseUntil = now + lease; @@ -28,6 +29,10 @@ private sealed record ClaimRow(Guid Id, string Kind, string Payload, int Attempt WHERE "Id" = ( SELECT "Id" FROM "WorkItems" WHERE "Status" = 'Queued' AND "Kind" = ANY({kinds}) + AND ("StorageConfigId" IS NULL + OR CAST({servedUserId} AS uuid) IS NULL + OR "StorageConfigId" IN ( + SELECT "Id" FROM "StorageConfigs" WHERE "UserId" = CAST({servedUserId} AS uuid))) ORDER BY "RequestedAt", "Id" LIMIT 1 FOR UPDATE SKIP LOCKED) diff --git a/src/Anichron.Worker/Work/IWorkItemStore.cs b/src/Anichron.Worker/Work/IWorkItemStore.cs index 3f7fcf3..afc5e8c 100644 --- a/src/Anichron.Worker/Work/IWorkItemStore.cs +++ b/src/Anichron.Worker/Work/IWorkItemStore.cs @@ -15,13 +15,17 @@ public sealed record ClaimedWork(WorkAttempt Attempt, WorkKind Kind, string Payl /// public interface IWorkItemStore { - /// The oldest queued item, now running under a lease, or null when none is queued. + /// + /// The oldest queued item this Worker may run, now running under a lease, or null when there is + /// none. Unscoped items go to any Worker; scoped ones only to a Worker serving their config. + /// /// The claiming Worker's display name. + /// A per-user Worker's user, or null for an all-user Worker. /// The claim time. /// How long the claim holds without a heartbeat. /// Cancels the claim. /// The claimed item, or null. - Task ClaimAsync(string claimedBy, Instant now, Duration lease, CancellationToken ct); + Task ClaimAsync(string claimedBy, Guid? servedUserId, Instant now, Duration lease, CancellationToken ct); /// Extends the lease and stores the progress snapshot. /// The running attempt. diff --git a/src/Anichron.Worker/Work/WorkExecutor.cs b/src/Anichron.Worker/Work/WorkExecutor.cs index 3fb72ae..4a17578 100644 --- a/src/Anichron.Worker/Work/WorkExecutor.cs +++ b/src/Anichron.Worker/Work/WorkExecutor.cs @@ -1,4 +1,5 @@ using Anichron.Core.Work; +using Anichron.Worker.Crawling; using Anichron.Worker.Scheduling; using Anichron.Worker.Settings; using Microsoft.Extensions.Options; @@ -9,6 +10,7 @@ namespace Anichron.Worker.Work; public sealed partial class WorkExecutor( IWorkItemStore store, WorkHandlerRegistry registry, + WorkerState workerState, IServiceScopeFactory scopeFactory, TimeProvider timeProvider, IOptions options, @@ -32,7 +34,7 @@ private async Task DrainOrLogAsync(CancellationToken stopping) { try { - while (await store.ClaimAsync(claimant, Now, Lease, stopping) is { } work) + while (await store.ClaimAsync(claimant, workerState.ResolvedUserId, Now, Lease, stopping) is { } work) await RunAsync(work, stopping); } catch (Exception ex) when (!stopping.IsCancellationRequested) @@ -125,7 +127,7 @@ private static partial class Log [LoggerMessage(Level = LogLevel.Information, Message = "Work item {Id} was handed back to the queue at shutdown.")] public static partial void Released(ILogger logger, Guid id); - [LoggerMessage(Level = LogLevel.Warning, Message = "Work item {Id} was recovered by another worker; this attempt stopped and wrote nothing more.")] + [LoggerMessage(Level = LogLevel.Warning, Message = "Work item {Id} is no longer held by this attempt (its lease was recovered, or the item was deleted with its storage config); this attempt stopped and wrote nothing more.")] public static partial void Fenced(ILogger logger, Guid id); [LoggerMessage(Level = LogLevel.Warning, Message = "Heartbeat for work item {Id} failed; will retry.")] diff --git a/src/Anichron.Worker/Work/WorkSchedule.cs b/src/Anichron.Worker/Work/WorkSchedule.cs index 6bebdd5..9bb0945 100644 --- a/src/Anichron.Worker/Work/WorkSchedule.cs +++ b/src/Anichron.Worker/Work/WorkSchedule.cs @@ -12,7 +12,13 @@ public interface IWorkSchedule Task EnqueueAsync(IWorkQueue queue, CancellationToken ct); } -public abstract class WorkSchedule(TPayload payload, Duration interval) : IWorkSchedule +/// The periodic schedules a Worker serves, asked again at every scheduler tick. +public interface IWorkScheduleSource +{ + Task> SchedulesAsync(IServiceProvider services, CancellationToken ct); +} + +public abstract class WorkSchedule(TPayload payload, Duration interval) : IWorkSchedule, IWorkScheduleSource where TPayload : IWorkPayload { public string DedupKey { get; } = payload.DedupKey @@ -21,4 +27,7 @@ public abstract class WorkSchedule(TPayload payload, Duration interval public Duration Interval => interval; public Task EnqueueAsync(IWorkQueue queue, CancellationToken ct) => queue.EnqueueAsync(payload, null, ct); + + public Task> SchedulesAsync(IServiceProvider services, CancellationToken ct) + => Task.FromResult>([this]); } diff --git a/src/Anichron.Worker/Work/WorkScheduler.cs b/src/Anichron.Worker/Work/WorkScheduler.cs index 54053b8..df7412a 100644 --- a/src/Anichron.Worker/Work/WorkScheduler.cs +++ b/src/Anichron.Worker/Work/WorkScheduler.cs @@ -8,7 +8,7 @@ namespace Anichron.Worker.Work; public sealed partial class WorkScheduler( IWorkItemStore store, - IEnumerable schedules, + IEnumerable sources, IServiceScopeFactory scopeFactory, TimeProvider timeProvider, IOptions options, @@ -36,13 +36,15 @@ private async Task TickAsync(CancellationToken ct) { var now = Instant.FromDateTimeOffset(timeProvider.GetUtcNow()); await store.RecoverExpiredLeasesAsync(now, settings.WorkMaxStalls, ct); - foreach (var schedule in schedules) + await using var scope = scopeFactory.CreateAsyncScope(); + var queue = scope.ServiceProvider.GetRequiredService(); + foreach (var source in sources) { - if (!await store.IsDueAsync(schedule.DedupKey, now - schedule.Interval, ct)) - continue; - - await using var scope = scopeFactory.CreateAsyncScope(); - await schedule.EnqueueAsync(scope.ServiceProvider.GetRequiredService(), ct); + foreach (var schedule in await source.SchedulesAsync(scope.ServiceProvider, ct)) + { + if (await store.IsDueAsync(schedule.DedupKey, now - schedule.Interval, ct)) + await schedule.EnqueueAsync(queue, ct); + } } }