From a48e648aece58a2e0a9874c427b7cd8c187cca1c Mon Sep 17 00:00:00 2001 From: bitbiter-dev Date: Mon, 19 Oct 2026 14:20:07 +1300 Subject: [PATCH 1/5] feat: Scope work items to a storage config and route claims by it (#281) - WorkItems.StorageConfigId, nullable, cascading on config delete, set from the payload's optional IWorkPayload.StorageConfigId. - The claim takes the Worker's served user: a per-user Worker claims unscoped items and items of its user's configs, an all-user Worker claims everything. The executor reads WorkerState.ResolvedUserId at each claim, after WorkerInitializer has resolved it. Contract tests on real PostgreSQL: per-user and all-user claims, a config delete removing its items and fencing the running one, and the scope stored on enqueue. Co-Authored-By: Claude Opus 5.5 --- docs/schema.sql | 6 + .../Work/WorkQueueTests.cs | 20 +- src/Anichron.Core/Data/AnichronDbContext.cs | 8 + src/Anichron.Core/Domain/WorkItem.cs | 1 + ...ScopeWorkItemsToStorageConfigs.Designer.cs | 692 ++++++++++++++++++ ...19011459_ScopeWorkItemsToStorageConfigs.cs | 51 ++ .../AnichronDbContextModelSnapshot.cs | 11 + src/Anichron.Core/Work/IWorkPayload.cs | 6 + src/Anichron.Core/Work/WorkQueue.cs | 1 + .../Work/MaintenanceHandlerTests.cs | 2 +- .../Work/WorkItemStoreTests.cs | 46 +- .../Work/WorkQueueTests.cs | 2 +- .../Work/WorkScopeTests.cs | 109 +++ .../Work/WorkExecutorTests.cs | 27 +- src/Anichron.Worker/Work/EfWorkItemStore.cs | 7 +- src/Anichron.Worker/Work/IWorkItemStore.cs | 8 +- src/Anichron.Worker/Work/WorkExecutor.cs | 4 +- 17 files changed, 967 insertions(+), 34 deletions(-) create mode 100644 src/Anichron.Core/Migrations/20261019011459_ScopeWorkItemsToStorageConfigs.Designer.cs create mode 100644 src/Anichron.Core/Migrations/20261019011459_ScopeWorkItemsToStorageConfigs.cs create mode 100644 src/Anichron.Tests.Integration/Work/WorkScopeTests.cs 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/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 775e2f7..b814edc 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/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/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/Work/WorkExecutorTests.cs b/src/Anichron.Worker.Tests.Unit/Work/WorkExecutorTests.cs index 0030e09..984ed04 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; @@ -33,6 +34,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 +42,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"); @@ -91,7 +93,7 @@ private async Task StartAsync(params ClaimedWork[] items) [WorkHandlerRegistration.For(), WorkHandlerRegistration.For()], [typeof(Cleanup), typeof(Retention)]); 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 +126,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 +156,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/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..fd2d006 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) From 6a79a2db9a2370eef1922d633b3be9015f77c82e Mon Sep 17 00:00:00 2001 From: bitbiter-dev Date: Mon, 19 Oct 2026 14:57:19 +1300 Subject: [PATCH 2/5] refactor(worker): Schedules come from sources asked at every tick (#281) A periodic kind can now have a varying set of schedules, such as one crawl per served storage config. WorkScheduler asks each IWorkScheduleSource for its schedules at every tick, inside the tick's scope; a fixed WorkSchedule is its own only source, so the token cleanup and retention schedules behave as before. Co-Authored-By: Claude Opus 5.5 --- .../WorkerServiceCollectionExtensionsTests.cs | 2 +- .../Work/WorkSchedulerTests.cs | 34 +++++++++++++++++-- .../WorkerServiceCollectionExtensions.cs | 4 +-- src/Anichron.Worker/Work/WorkSchedule.cs | 11 +++++- src/Anichron.Worker/Work/WorkScheduler.cs | 16 +++++---- 5 files changed, 54 insertions(+), 13 deletions(-) diff --git a/src/Anichron.Worker.Tests.Unit/Infrastructure/WorkerServiceCollectionExtensionsTests.cs b/src/Anichron.Worker.Tests.Unit/Infrastructure/WorkerServiceCollectionExtensionsTests.cs index fd7b6cc..29ca558 100644 --- a/src/Anichron.Worker.Tests.Unit/Infrastructure/WorkerServiceCollectionExtensionsTests.cs +++ b/src/Anichron.Worker.Tests.Unit/Infrastructure/WorkerServiceCollectionExtensionsTests.cs @@ -231,7 +231,7 @@ public void AddWorkQueueServices_RegistersEachKindsHandlerAsScoped() [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), 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/Infrastructure/WorkerServiceCollectionExtensions.cs b/src/Anichron.Worker/Infrastructure/WorkerServiceCollectionExtensions.cs index 0bf9471..1a81d7b 100644 --- a/src/Anichron.Worker/Infrastructure/WorkerServiceCollectionExtensions.cs +++ b/src/Anichron.Worker/Infrastructure/WorkerServiceCollectionExtensions.cs @@ -58,8 +58,8 @@ internal IServiceCollection AddWorkQueueServices() services.AddSingleton(); services.AddWorkHandler(); services.AddWorkHandler(); - services.AddSingleton(); - services.AddSingleton(); + services.AddSingleton(); + services.AddSingleton(); return services; } 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); + } } } From fee574897b5a6b1597e14dc9e391787251da6746 Mon Sep 17 00:00:00 2001 From: bitbiter-dev Date: Mon, 19 Oct 2026 20:06:20 +1300 Subject: [PATCH 3/5] feat(worker): Crawl each storage config as a work item (#281) - WorkKind.Crawl with CrawlPayload(StorageConfigId): dedup key crawl:, scoped to its config. - CrawlSchedules: one schedule per active config the Worker serves (its user's configs, or all of them for an all-user Worker), due every Worker:CrawlIntervalHours. The served user is read at each tick. The first crawl after a start comes from due-ness. - CrawlHandler runs today's crawl through FileIngestionPipeline, unchanged in concurrency, exclusions and failure handling. A missing or inactive config is skipped; a missing root fails only that config's item (#282); a shutdown propagates and the item is released. - The pipeline counts items found, processed and failed (a Live Photo pair is one item) under one lock and reports them as progress, with no total until the walk ends. The result is the final counts. - The crawl BackgroundService and its timer are gone. Co-Authored-By: Claude Opus 5.5 --- CLAUDE.md | 13 +- .../Work/PayloadsTests.cs | 17 ++ src/Anichron.Core/Domain/WorkKind.cs | 1 + src/Anichron.Core/Work/Payloads.cs | 9 + .../Crawling/CrawlTests.cs | 152 ++++++++++ .../Crawling/FileIngestionPipelineTests.cs | 105 +++++-- .../Crawling/WorkerTests.cs | 264 ------------------ .../WorkerServiceCollectionExtensionsTests.cs | 9 +- .../Work/WorkExecutorTests.cs | 9 +- .../Work/WorkHandlerRegistryTests.cs | 24 +- src/Anichron.Worker/Crawling/Crawl.cs | 54 ++++ src/Anichron.Worker/Crawling/CrawlTally.cs | 47 ++++ .../Crawling/FileIngestionPipeline.cs | 26 +- src/Anichron.Worker/Crawling/Worker.cs | 67 ----- src/Anichron.Worker/Crawling/WorkerState.cs | 4 +- .../WorkerServiceCollectionExtensions.cs | 4 +- 16 files changed, 425 insertions(+), 380 deletions(-) create mode 100644 src/Anichron.Worker.Tests.Unit/Crawling/CrawlTests.cs delete mode 100644 src/Anichron.Worker.Tests.Unit/Crawling/WorkerTests.cs create mode 100644 src/Anichron.Worker/Crawling/Crawl.cs create mode 100644 src/Anichron.Worker/Crawling/CrawlTally.cs delete mode 100644 src/Anichron.Worker/Crawling/Worker.cs diff --git a/CLAUDE.md b/CLAUDE.md index d181b9d..fcd7d51 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 @@ -193,12 +193,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/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/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/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.Worker.Tests.Unit/Crawling/CrawlTests.cs b/src/Anichron.Worker.Tests.Unit/Crawling/CrawlTests.cs new file mode 100644 index 0000000..82562bb --- /dev/null +++ b/src/Anichron.Worker.Tests.Unit/Crawling/CrawlTests.cs @@ -0,0 +1,152 @@ +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 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, logger).RunAsync(new CrawlPayload(configId), progress, Ct); + + // ========================================================================== + // CrawlHandler + // ========================================================================== + + [Fact] + public async Task RunAsync_ActiveConfig_CrawlsItAndReturnsTheCountsAsync() + { + var config = Config(); + configs.FindByIdAsync(config.Id, Ct).Returns(config); + pipeline.RunAsync(config, progress, Ct).Returns(new CrawlResult(5, 4, 1)); + + var result = await RunAsync(config.Id); + + result.Should().Be(new CrawlResult(5, 4, 1)); + logger.Collector.GetSnapshot().Should().ContainSingle(e => + e.Level == LogLevel.Information && e.Message.Contains("/data/originals/mira")); + } + + [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(); + } + + [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/WorkerTests.cs b/src/Anichron.Worker.Tests.Unit/Crawling/WorkerTests.cs deleted file mode 100644 index 9f6559c..0000000 --- a/src/Anichron.Worker.Tests.Unit/Crawling/WorkerTests.cs +++ /dev/null @@ -1,264 +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 WorkerState WorkerState { get; } = new(); - - private readonly IServiceScopeFactory scopeFactory; - - public TestFixture() - { - scopeFactory = ScopeFactories.Providing(services => services.AddSingleton(ConfigRepository)); - - 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()); - 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()); - 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(); - } - - [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 29ca558..ef87483 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,6 +224,7 @@ public void AddWorkQueueServices_RegistersEachKindsHandlerAsScoped() .Select(d => (d.ServiceType, d.ImplementationType)).Should().BeEquivalentTo([ (typeof(IWorkHandler), typeof(TokenCleanupHandler)), (typeof(IWorkHandler), typeof(WorkItemRetentionHandler)), + (typeof(IWorkHandler), typeof(CrawlHandler)), ]); } @@ -235,6 +235,7 @@ public void AddWorkQueueServices_RegistersEachPeriodicKindsSchedule() .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 984ed04..5570d2d 100644 --- a/src/Anichron.Worker.Tests.Unit/Work/WorkExecutorTests.cs +++ b/src/Anichron.Worker.Tests.Unit/Work/WorkExecutorTests.cs @@ -25,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); @@ -90,8 +95,8 @@ 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, workerState, ScopeFactories.Providing(services => services.AddSingleton(handler)), timeline.Clock, Options.Create(new WorkerSettings()), logger); 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/Crawling/Crawl.cs b/src/Anichron.Worker/Crawling/Crawl.cs new file mode 100644 index 0000000..19bb71f --- /dev/null +++ b/src/Anichron.Worker/Crawling/Crawl.cs @@ -0,0 +1,54 @@ +using Anichron.Core.Data.Repository; +using Anichron.Core.Work; +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, + 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); + return await pipeline.RunAsync(config, progress, ct); + } + + 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.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..3168d58 --- /dev/null +++ b/src/Anichron.Worker/Crawling/CrawlTally.cs @@ -0,0 +1,47 @@ +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); + +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/Worker.cs b/src/Anichron.Worker/Crawling/Worker.cs deleted file mode 100644 index 6ba7fe5..0000000 --- a/src/Anichron.Worker/Crawling/Worker.cs +++ /dev/null @@ -1,67 +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); - } - - 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 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); - } -} 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 1a81d7b..7fb500a 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.AddWorkHandler(); services.AddSingleton(); services.AddSingleton(); + services.AddSingleton(); return services; } @@ -121,7 +122,6 @@ internal IServiceCollection AddWorkerHostedServices() services.AddHostedService(); services.AddHostedService(); services.AddHostedService(); - services.AddHostedService(); return services; } } From 0d32955f0dfdd5a4a6bf96b04e7293a20875d2f3 Mon Sep 17 00:00:00 2001 From: bitbiter-dev Date: Tue, 20 Oct 2026 14:09:04 +1300 Subject: [PATCH 4/5] fix(worker): Name both reasons an attempt loses its item in the fenced log (#281) Deleting a storage config mid-crawl deletes its running item, so the heartbeat is fenced without any other worker involved. Co-Authored-By: Claude Opus 5.5 --- src/Anichron.Worker/Work/WorkExecutor.cs | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/src/Anichron.Worker/Work/WorkExecutor.cs b/src/Anichron.Worker/Work/WorkExecutor.cs index fd2d006..4a17578 100644 --- a/src/Anichron.Worker/Work/WorkExecutor.cs +++ b/src/Anichron.Worker/Work/WorkExecutor.cs @@ -127,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.")] From ca0bb0b3438d19a975b23a91e350700b948b3e9a Mon Sep 17 00:00:00 2001 From: bitbiter-dev Date: Tue, 20 Oct 2026 15:06:52 +1300 Subject: [PATCH 5/5] test(worker): Kill the two surviving mutants in the duplicate report (#281) - The queried config ids are a Union, so their order is pinned by a test instead of being an equivalent Append/Prepend mutant. - A crawl with pairs but no copies within the config is summarised. Co-Authored-By: Claude Opus 5.5 --- .../Crawling/SharedContentReporterTests.cs | 4 +++- src/Anichron.Worker/Crawling/SharedContentReporter.cs | 2 +- 2 files changed, 4 insertions(+), 2 deletions(-) diff --git a/src/Anichron.Worker.Tests.Unit/Crawling/SharedContentReporterTests.cs b/src/Anichron.Worker.Tests.Unit/Crawling/SharedContentReporterTests.cs index bb5de56..3d92671 100644 --- a/src/Anichron.Worker.Tests.Unit/Crawling/SharedContentReporterTests.cs +++ b/src/Anichron.Worker.Tests.Unit/Crawling/SharedContentReporterTests.cs @@ -52,7 +52,7 @@ public async Task ReportAsync_LooksAcrossTheCrawledConfigAndTheUsersOtherConfigs await fx.ReportAsync(Phone, [Home, Backup]); - fx.QueriedConfigs.Should().BeEquivalentTo([Home, Backup, Phone]); + fx.QueriedConfigs.Should().Equal(Home, Backup, Phone); } [Fact] @@ -76,6 +76,8 @@ public async Task ReportAsync_ContentInTwoConfigs_LogsThePairOnceAtDebugAsync() new ContentLocation(Mira, "cafe", Home, "2023/a.jpg")); 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 " diff --git a/src/Anichron.Worker/Crawling/SharedContentReporter.cs b/src/Anichron.Worker/Crawling/SharedContentReporter.cs index a44f890..a2814d1 100644 --- a/src/Anichron.Worker/Crawling/SharedContentReporter.cs +++ b/src/Anichron.Worker/Crawling/SharedContentReporter.cs @@ -24,7 +24,7 @@ public async Task ReportAsync( UserStorageConfig crawled, IReadOnlyCollection usersConfigs, CancellationToken ct) { var copies = tally.Take(crawled.Id); - var configIds = usersConfigs.Select(config => config.Id).Append(crawled.Id).Distinct().ToList(); + var configIds = usersConfigs.Select(config => config.Id).Union([crawled.Id]).ToList(); var locations = await repository.FindSharedContentAsync(configIds, ct); var pairs = 0;