From 6cc169e69a5b8ed0f3c358ffe9ef2fd42646b997 Mon Sep 17 00:00:00 2001 From: bitbiter-dev Date: Wed, 21 Oct 2026 13:25:13 +1300 Subject: [PATCH 1/4] feat: Cancel queued and running work items (#280) POST /api/v1/admin/work-items/{id}/cancel cancels a queued item at once (200) or records CancelRequestedAt on a running one (202); a finished item is 409 and an unknown id 404. The write is one status-guarded UPDATE, so a cancel racing a claim either cancels the queued item or reaches the running one. The heartbeat reports the request, the executor cancels the handler and keeps the lease until it returns, and the item ends Cancelled with its last progress, shown as partial. Release and lease recovery never requeue a requested item. Architecture rule api-writes-work-items-only-by-cancel keeps every other work-item write in the Worker. Co-Authored-By: Claude Opus 5.5 --- CLAUDE.md | 7 +- ...ope-excludes-unkillable-and-wiring-code.md | 1 + docs/schema.sql | 6 +- .../Endpoints/WorkItemEndpointsHttpTests.cs | 33 +- .../Endpoints/WorkItemEndpointsTests.cs | 108 ++- .../ApiServiceRegistrationTests.cs | 1 + .../Infrastructure/RouteTableTests.cs | 12 +- src/Anichron.API/Endpoints/WorkItemDto.cs | 2 + .../Endpoints/WorkItemDtoMapper.cs | 4 +- .../Endpoints/WorkItemEndpoints.cs | 29 + src/Anichron.API/Infrastructure/ApiPaths.cs | 4 + .../ServiceCollectionExtensions.cs | 1 + .../ArchitectureRules.cs | 10 + .../Work/WorkItemReaderTests.cs | 4 +- src/Anichron.Core/Data/AnichronDbContext.cs | 12 +- src/Anichron.Core/Domain/WorkItem.cs | 1 + ...20261020234633_CancelWorkItems.Designer.cs | 697 ++++++++++++++++++ .../20261020234633_CancelWorkItems.cs | 62 ++ .../AnichronDbContextModelSnapshot.cs | 8 +- src/Anichron.Core/Work/WorkItemCanceller.cs | 46 ++ src/Anichron.Core/Work/WorkItemReader.cs | 6 +- .../Work/WorkDatabase.cs | 6 + .../Work/WorkItemCancellerTests.cs | 122 +++ .../Work/WorkItemStoreTests.cs | 91 ++- .../Work/WorkScopeTests.cs | 3 +- .../Work/WorkExecutorTests.cs | 130 +++- src/Anichron.Worker/Work/EfWorkItemStore.cs | 29 +- src/Anichron.Worker/Work/IWorkItemStore.cs | 32 +- src/Anichron.Worker/Work/WorkExecutor.cs | 26 +- 29 files changed, 1442 insertions(+), 51 deletions(-) create mode 100644 src/Anichron.Core/Migrations/20261020234633_CancelWorkItems.Designer.cs create mode 100644 src/Anichron.Core/Migrations/20261020234633_CancelWorkItems.cs create mode 100644 src/Anichron.Core/Work/WorkItemCanceller.cs create mode 100644 src/Anichron.Tests.Integration/Work/WorkItemCancellerTests.cs diff --git a/CLAUDE.md b/CLAUDE.md index 8094eeb..bdc9325 100644 --- a/CLAUDE.md +++ b/CLAUDE.md @@ -100,7 +100,7 @@ Four projects with strict separation of concerns: **Write ownership is split by domain, not centralized.** - **Worker is the only writer of *media* data** — `MediaAsset`, `Metadata`, `ProxyFile`, `Burst`, plus all reconciliation and soft-deletes happen here. The API never writes media: outside its DI wiring it never touches `AnichronDbContext` (architecture rule `api-reaches-data-through-repositories`), and the media write repository stays out of it (`only-worker-writes-media`). Its one identity-side write is the `TokenCleanup` work item's handler deleting expired refresh tokens. - **API owns all *identity/auth/admin* writes** — registration, login/refresh-token issuance and revocation, lockout state, invites, and admin user/storage-config CRUD all persist through API services (`AuthService`, `AdminUserService`, `AdminStorageConfigService`, `TokenService`, etc.) via `IUnitOfWork.SaveChangesAsync`. The API is read-only *only with respect to media*. The one exception is `AssetInteraction` (star/like/hide, #46): per-user state, written by the API through `IAssetInteractionWriter` (Core), which takes and returns plain records so no API type touches the entity. It checks only `VisibleTo` (never `NotHiddenFor`, so a hidden asset can be unhidden), and a first write that loses the `(UserId, AssetId)` unique race reloads and reapplies once. -- **Work items** are inserted only through `IWorkQueue` (architecture rule `only-work-queue-inserts-work-items`); every later state change belongs to the Worker's executor, and handlers live only in the Worker (`work-handlers-live-in-worker`). See "Work queue" below. +- **Work items** are inserted only through `IWorkQueue` (architecture rule `only-work-queue-inserts-work-items`); the API's only other write is an admin's cancel through `IWorkItemCanceller` (`api-writes-work-items-only-by-cancel`: outside the Worker, only `EfWorkQueue`, `EfWorkItemReader` and `EfWorkItemCanceller` touch `WorkItems`); every other state change belongs to the Worker's executor, and handlers live only in the Worker (`work-handlers-live-in-worker`). See "Work queue" below. - Both hosts apply EF migrations at startup, serialised by a PostgreSQL advisory lock (`MigrateWithAdvisoryLockAsync`). A planned `Anichron.UI` (NextJS/React) for the Instagram Story-style flashback experience is documented but not yet in the solution. @@ -199,8 +199,9 @@ Durable work for the Worker lives in `WorkItems` (#276, ADR 0008). Its kinds are - **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). - **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. `UpdatedAt` is the change cursor, so a heartbeat moves it only when the snapshot changed; `LeaseUntil` is what says the Worker is alive. 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. -- **Admin view (#279):** `GET /api/v1/admin/work-items` (filtered by status and kind, newest request first, paged), `.../{id}`, and `.../events`, a Server-Sent Events stream that re-queries `UpdatedAt` every `WorkItemEvents:IntervalSeconds` (2 s). An event's id is the cursor, and the list carries the cursor read before its page, so a client that lists and then subscribes with it as `Last-Event-ID` misses nothing; an item may arrive twice and is merged by id. The API reads items only through `IWorkItemReader` (Core), as `WorkItemView`. A result is shown once an item succeeded, an error once it failed or was cancelled. +- **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. `UpdatedAt` is the change cursor, so a heartbeat moves it only when the snapshot changed; `LeaseUntil` is what says the Worker is alive. 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. +- **Cancelling:** `POST /api/v1/admin/work-items/{id}/cancel` (#280) goes through `EfWorkItemCanceller`, one status-guarded `UPDATE`: a queued item ends `Cancelled` at once (`200`), a running one gets `CancelRequestedAt` (`202`), a finished one is `409` and an unknown id `404`. The heartbeat reads the request and cancels the handler but keeps beating until it returns; the item then ends `Cancelled` with its last progress snapshot (and a summary, if the handler returned one), shown as partial (`ResultIsPartial`). A handler that fails after the request still fails. A requested item is never queued again: release and lease recovery end it `Cancelled`, even at the stall limit. 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. +- **Admin view (#279):** `GET /api/v1/admin/work-items` (filtered by status and kind, newest request first, paged), `.../{id}`, and `.../events`, a Server-Sent Events stream that re-queries `UpdatedAt` every `WorkItemEvents:IntervalSeconds` (2 s). An event's id is the cursor, and the list carries the cursor read before its page, so a client that lists and then subscribes with it as `Last-Event-ID` misses nothing; an item may arrive twice and is merged by id. The API reads items only through `IWorkItemReader` (Core), as `WorkItemView`. A result is shown once an item succeeded or was cancelled (then marked partial), an error once it failed or was cancelled. - **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`. diff --git a/docs/adr/0005-mutation-scope-excludes-unkillable-and-wiring-code.md b/docs/adr/0005-mutation-scope-excludes-unkillable-and-wiring-code.md index 5d37283..7e06dcc 100644 --- a/docs/adr/0005-mutation-scope-excludes-unkillable-and-wiring-code.md +++ b/docs/adr/0005-mutation-scope-excludes-unkillable-and-wiring-code.md @@ -87,6 +87,7 @@ The directive is left bare — the reason lives here, not in the code. | `AnichronDbContext.ExecuteInTransactionAsync` (both overloads) | Commit versus no commit is only observable on a provider with real transactions. | | `EfRefreshTokenRepository.RevokeAllActiveByUserIdAsync` and `DeleteExpiredAsync` | `ExecuteUpdate`/`ExecuteDelete` are not supported by the EF InMemory provider. | | `EfWorkQueue.TryInsertAsync` | Absorbing a lost race needs the partial unique index on `DedupKey`, which the InMemory provider does not enforce. The rest of `EnqueueAsync` is in scope. | +| `EfWorkItemCanceller.CancelAsync` | One raw `UPDATE … RETURNING` guarded by status, so a cancel racing a claim either cancels the queued item or records the request on the running one; only PostgreSQL's row locking decides that race. Covered by `WorkItemCancellerTests` on real PostgreSQL. | | `EfWorkItemStore` members except `IsDueAsync`, and its `Running`/`Expired` filters | The claim is raw SQL with `FOR UPDATE SKIP LOCKED`; every transition is an `ExecuteUpdate`/`ExecuteDelete` fenced by `(Id, Attempt)`. `IsDueAsync` is plain LINQ and in scope. | These are **not** tested against a second SQL engine such as SQLite. A substitute engine diff --git a/docs/schema.sql b/docs/schema.sql index 435cf7c..1365bd9 100644 --- a/docs/schema.sql +++ b/docs/schema.sql @@ -75,6 +75,7 @@ CREATE TABLE "WorkItems" ( "LeaseUntil" timestamp with time zone, "FinishedAt" timestamp with time zone, "UpdatedAt" timestamp with time zone NOT NULL, + "CancelRequestedAt" timestamp with time zone, "ProgressDone" bigint NOT NULL, "ProgressTotal" bigint, "ProgressNote" text, @@ -84,7 +85,7 @@ CREATE TABLE "WorkItems" ( 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 TABLE "WorkItems" IS 'Durable work for the Worker: periodic maintenance today, admin requests later. Only IWorkQueue inserts rows; the API''s only other write is an admin''s cancel, and every other 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.'; @@ -96,8 +97,9 @@ COMMENT ON COLUMN "WorkItems"."Stalls" IS 'Lease expiries, counted by lease reco COMMENT ON COLUMN "WorkItems"."ClaimedBy" IS 'The claiming Worker for display: its WORKER__USER or ''all'', then its host name.'; COMMENT ON COLUMN "WorkItems"."LeaseUntil" IS 'Extended by every heartbeat. A Running item past it belongs to a dead or paused Worker and goes back to the queue.'; COMMENT ON COLUMN "WorkItems"."UpdatedAt" IS 'Moves on every insert, claim and transition, and on a heartbeat only when the progress snapshot changed, so a reader can ask what changed since a point in time.'; +COMMENT ON COLUMN "WorkItems"."CancelRequestedAt" IS 'When an admin asked to cancel the item. A Queued item is cancelled at once; a Running one keeps running until the Worker''s next heartbeat reads this and cancels its handler. A requested item is never queued again: release and lease recovery end it Cancelled.'; COMMENT ON COLUMN "WorkItems"."ProgressNote" IS 'The latest progress snapshot (done, optional total, note), overwritten by each heartbeat; no history is kept.'; -COMMENT ON COLUMN "WorkItems"."Result" IS 'The handler''s results summary, written when the item succeeds.'; +COMMENT ON COLUMN "WorkItems"."Result" IS 'The handler''s results summary, written when the item succeeds, or when it is cancelled after its handler returned one; a Cancelled item''s results are partial.'; CREATE TABLE "Bursts" ( diff --git a/src/Anichron.API.Tests.Unit/Endpoints/WorkItemEndpointsHttpTests.cs b/src/Anichron.API.Tests.Unit/Endpoints/WorkItemEndpointsHttpTests.cs index 85950cb..677ff7e 100644 --- a/src/Anichron.API.Tests.Unit/Endpoints/WorkItemEndpointsHttpTests.cs +++ b/src/Anichron.API.Tests.Unit/Endpoints/WorkItemEndpointsHttpTests.cs @@ -19,6 +19,7 @@ public sealed class WorkItemEndpointsHttpTests private static readonly Guid ItemId = Guid.Parse("aaaaaaaa-aaaa-aaaa-aaaa-aaaaaaaaaaaa"); private readonly IWorkItemReader reader = Substitute.For(); + private readonly IWorkItemCanceller canceller = Substitute.For(); private static CancellationToken Ct => TestContext.Current.CancellationToken; @@ -31,6 +32,7 @@ private Task StartAsync() => AuthenticatedTestApp.StartAsync(api => api.MapWorkItemEndpoints(), services => { services.AddSingleton(reader); + services.AddSingleton(canceller); services.AddSingleton(new FakeTimeProvider()); services.AddSingleton(Options.Create(new WorkItemEventsSettings())); services.AddScoped(); @@ -69,17 +71,38 @@ public async Task List_UnknownStatus_IsABadRequestAsync() } [Theory] - [InlineData("/api/v1/admin/work-items")] - [InlineData("/api/v1/admin/work-items/aaaaaaaa-aaaa-aaaa-aaaa-aaaaaaaaaaaa")] - [InlineData("/api/v1/admin/work-items/events")] - public async Task EveryRoute_ForANonAdmin_IsForbiddenAsync(string path) + [InlineData("GET", "/api/v1/admin/work-items")] + [InlineData("GET", "/api/v1/admin/work-items/aaaaaaaa-aaaa-aaaa-aaaa-aaaaaaaaaaaa")] + [InlineData("GET", "/api/v1/admin/work-items/events")] + [InlineData("POST", "/api/v1/admin/work-items/aaaaaaaa-aaaa-aaaa-aaaa-aaaaaaaaaaaa/cancel")] + public async Task EveryRoute_ForANonAdmin_IsForbiddenAsync(string method, string path) { await using var app = await StartAsync(); using var client = app.Client; + using var request = AuthenticatedTestApp.Get(path, AdminId); + request.Method = new HttpMethod(method); - using var response = await client.SendAsync(AuthenticatedTestApp.Get(path, AdminId), Ct); + using var response = await client.SendAsync(request, Ct); response.StatusCode.Should().Be(HttpStatusCode.Forbidden); + await canceller.DidNotReceive().CancelAsync(Arg.Any(), Arg.Any()); + } + + [Fact] + public async Task Cancel_PostedByAnAdmin_CancelsTheItemInThePathAsync() + { + canceller.CancelAsync(ItemId, Arg.Any()).Returns(CancelOutcome.Finished); + reader.FindAsync(ItemId, Arg.Any()).Returns(Succeeded); + await using var app = await StartAsync(); + using var client = app.Client; + using var request = AuthenticatedTestApp.GetAsAdmin($"/api/v1/admin/work-items/{ItemId}/cancel", AdminId); + request.Method = HttpMethod.Post; + + using var response = await client.SendAsync(request, Ct); + + response.StatusCode.Should().Be(HttpStatusCode.Conflict); + using var body = JsonDocument.Parse(await response.Content.ReadAsStringAsync(Ct)); + body.RootElement.GetProperty("status").GetString().Should().Be("Succeeded"); } [Fact] diff --git a/src/Anichron.API.Tests.Unit/Endpoints/WorkItemEndpointsTests.cs b/src/Anichron.API.Tests.Unit/Endpoints/WorkItemEndpointsTests.cs index e3a4b4b..d4d04fb 100644 --- a/src/Anichron.API.Tests.Unit/Endpoints/WorkItemEndpointsTests.cs +++ b/src/Anichron.API.Tests.Unit/Endpoints/WorkItemEndpointsTests.cs @@ -15,6 +15,7 @@ public sealed class WorkItemEndpointsTests private static readonly Guid AdminId = Guid.Parse("dddddddd-dddd-dddd-dddd-dddddddddddd"); private readonly IWorkItemReader reader = Substitute.For(); + private readonly IWorkItemCanceller canceller = Substitute.For(); private static CancellationToken Ct => TestContext.Current.CancellationToken; @@ -151,6 +152,73 @@ public async Task GetAsync_UnknownId_IsNotFoundAsync() // StreamEvents // ========================================================================== + // ========================================================================== + // CancelAsync + // ========================================================================== + + private void Cancels(CancelOutcome outcome, WorkItemStatus statusAfter) + { + canceller.CancelAsync(ItemId, Arg.Any()).Returns(outcome); + reader.FindAsync(ItemId, Arg.Any()).Returns(View(statusAfter)); + } + + [Fact] + public async Task CancelAsync_QueuedItem_ReturnsTheCancelledItemAsync() + { + Cancels(CancelOutcome.Cancelled, WorkItemStatus.Cancelled); + + var result = await WorkItemEndpoints.CancelAsync(ItemId, canceller, reader, Ct); + + result.Result.Should().BeOfType>().Which.Value!.Status.Should().Be("Cancelled"); + } + + [Fact] + public async Task CancelAsync_RunningItem_IsAcceptedWithTheItemAndItsLocationAsync() + { + Cancels(CancelOutcome.Requested, WorkItemStatus.Running); + + var result = await WorkItemEndpoints.CancelAsync(ItemId, canceller, reader, Ct); + + var accepted = result.Result.Should().BeOfType>().Which; + accepted.Location.Should().Be("/api/v1/admin/work-items/aaaaaaaa-aaaa-aaaa-aaaa-aaaaaaaaaaaa"); + accepted.Value!.Status.Should().Be("Running"); + } + + [Fact] + public async Task CancelAsync_FinishedItem_IsAConflictShowingTheItemAsync() + { + Cancels(CancelOutcome.Finished, WorkItemStatus.Succeeded); + + var result = await WorkItemEndpoints.CancelAsync(ItemId, canceller, reader, Ct); + + result.Result.Should().BeOfType>().Which.Value!.Status.Should().Be("Succeeded"); + } + + [Fact] + public async Task CancelAsync_UnknownId_IsNotFoundWithoutReadingAsync() + { + canceller.CancelAsync(ItemId, Arg.Any()).Returns(CancelOutcome.NotFound); + + var result = await WorkItemEndpoints.CancelAsync(ItemId, canceller, reader, Ct); + + result.Result.Should().BeOfType(); + await reader.DidNotReceive().FindAsync(Arg.Any(), Arg.Any()); + } + + [Theory] + [InlineData(CancelOutcome.Cancelled)] + [InlineData(CancelOutcome.Requested)] + [InlineData(CancelOutcome.Finished)] + public async Task CancelAsync_ItemDeletedBeforeItIsRead_IsNotFoundAsync(CancelOutcome outcome) + { + canceller.CancelAsync(ItemId, Arg.Any()).Returns(outcome); + reader.FindAsync(ItemId, Arg.Any()).Returns((WorkItemView?)null); + + var result = await WorkItemEndpoints.CancelAsync(ItemId, canceller, reader, Ct); + + result.Result.Should().BeOfType(); + } + [Fact] public void StreamEvents_AnswersWithAServerSentEventStream() { @@ -208,18 +276,47 @@ public void ToDto_Succeeded_ShowsTheResultAsJsonAndNoError() public void ToDto_SucceededWithoutAResult_ShowsNoResult() => WorkItemDtoMapper.ToDto(View(WorkItemStatus.Succeeded, result: null)).Result.Should().BeNull(); - [Theory] - [InlineData(WorkItemStatus.Failed)] - [InlineData(WorkItemStatus.Cancelled)] - public void ToDto_FailedOrCancelled_ShowsTheErrorAndProgressButNoResult(WorkItemStatus status) + [Fact] + public void ToDto_Succeeded_ResultIsNotPartial() + => WorkItemDtoMapper.ToDto(View(WorkItemStatus.Succeeded)).ResultIsPartial.Should().BeFalse(); + + [Fact] + public void ToDto_Failed_ShowsTheErrorAndProgressButNoResult() { - var dto = WorkItemDtoMapper.ToDto(View(status)); + var dto = WorkItemDtoMapper.ToDto(View(WorkItemStatus.Failed)); dto.Error.Should().Be("IOException: the root is gone"); dto.Progress.Should().Be(new WorkProgressDto(7, 8, "7 found")); dto.Result.Should().BeNull(); + dto.ResultIsPartial.Should().BeFalse(); } + [Fact] + public void ToDto_Cancelled_ShowsTheResultAndProgressMarkedPartial() + { + var dto = WorkItemDtoMapper.ToDto(View(WorkItemStatus.Cancelled, error: null)); + + dto.Result.Should().NotBeNull(); + dto.Result.Value.GetProperty("found").GetInt32().Should().Be(8); + dto.Progress.Should().Be(new WorkProgressDto(7, 8, "7 found")); + dto.ResultIsPartial.Should().BeTrue(); + dto.Error.Should().BeNull(); + } + + [Fact] + public void ToDto_CancelledAsSuperseded_ShowsTheError() + => WorkItemDtoMapper.ToDto(View(WorkItemStatus.Cancelled, result: null)).Error + .Should().Be("IOException: the root is gone"); + + [Fact] + public void ToDto_CancelRequested_ShowsWhen() + => WorkItemDtoMapper.ToDto(View(WorkItemStatus.Running) with { CancelRequestedAt = At(4) }).CancelRequestedAt + .Should().Be("2026-10-18T09:00:04Z"); + + [Fact] + public void ToDto_NoCancelRequested_ShowsNone() + => WorkItemDtoMapper.ToDto(View(WorkItemStatus.Running)).CancelRequestedAt.Should().BeNull(); + [Theory] [InlineData(WorkItemStatus.Queued)] [InlineData(WorkItemStatus.Running)] @@ -229,5 +326,6 @@ public void ToDto_NotFinished_ShowsNeitherResultNorError(WorkItemStatus status) dto.Result.Should().BeNull(); dto.Error.Should().BeNull(); + dto.ResultIsPartial.Should().BeFalse(); } } diff --git a/src/Anichron.API.Tests.Unit/Infrastructure/ApiServiceRegistrationTests.cs b/src/Anichron.API.Tests.Unit/Infrastructure/ApiServiceRegistrationTests.cs index ac2a541..8d9e8ec 100644 --- a/src/Anichron.API.Tests.Unit/Infrastructure/ApiServiceRegistrationTests.cs +++ b/src/Anichron.API.Tests.Unit/Infrastructure/ApiServiceRegistrationTests.cs @@ -119,6 +119,7 @@ private static ServiceCollection WorkItemServices(Dictionary? v [Theory] [InlineData(typeof(IWorkItemReader), typeof(EfWorkItemReader))] + [InlineData(typeof(IWorkItemCanceller), typeof(EfWorkItemCanceller))] [InlineData(typeof(WorkItemEvents), typeof(WorkItemEvents))] public void AddWorkItemServices_RegistersTheScopedImplementation(Type service, Type implementation) => WorkItemServices().Should().ContainSingle(d => d.ServiceType == service) diff --git a/src/Anichron.API.Tests.Unit/Infrastructure/RouteTableTests.cs b/src/Anichron.API.Tests.Unit/Infrastructure/RouteTableTests.cs index b474679..553d918 100644 --- a/src/Anichron.API.Tests.Unit/Infrastructure/RouteTableTests.cs +++ b/src/Anichron.API.Tests.Unit/Infrastructure/RouteTableTests.cs @@ -144,8 +144,14 @@ public void FlashbackRoutes_AreTaggedFlashbacks() [InlineData("/api/v1/admin/work-items/{workItemId:guid}")] [InlineData("/api/v1/admin/work-items/events")] public void WorkItemRoutes_RequireTheAdminPolicyWithoutRateLimit(string pattern) + => AssertAdminWithoutRateLimit(Find("GET", pattern)); + + [Fact] + public void CancelWorkItemRoute_RequiresTheAdminPolicyWithoutRateLimit() + => AssertAdminWithoutRateLimit(Find("POST", "/api/v1/admin/work-items/{workItemId:guid}/cancel")); + + private static void AssertAdminWithoutRateLimit(Route route) { - var route = Find("GET", pattern); route.Anonymous.Should().BeFalse(); route.Policies.Should().Contain(AuthPolicies.Admin); @@ -155,7 +161,7 @@ public void WorkItemRoutes_RequireTheAdminPolicyWithoutRateLimit(string pattern) [Fact] public void WorkItemRoutes_AreTaggedWorkItems() => Routes.Where(r => r.Pattern.StartsWith("/api/v1/admin/work-items", StringComparison.Ordinal)) - .Should().HaveCount(3).And.OnlyContain(r => r.Tags.SequenceEqual(new[] { "Work items" })); + .Should().HaveCount(4).And.OnlyContain(r => r.Tags.SequenceEqual(new[] { "Work items" })); [Fact] public void Healthz_IsAnonymous() @@ -163,5 +169,5 @@ public void Healthz_IsAnonymous() [Fact] public void EveryMappedRoute_IsCoveredByTheTestsAbove() - => Routes.Should().HaveCount(25); + => Routes.Should().HaveCount(26); } diff --git a/src/Anichron.API/Endpoints/WorkItemDto.cs b/src/Anichron.API/Endpoints/WorkItemDto.cs index cfc5885..c15bdeb 100644 --- a/src/Anichron.API/Endpoints/WorkItemDto.cs +++ b/src/Anichron.API/Endpoints/WorkItemDto.cs @@ -15,8 +15,10 @@ public sealed record WorkItemDto( string? StartedAt, string? FinishedAt, string UpdatedAt, + string? CancelRequestedAt, WorkProgressDto Progress, JsonElement? Result, + bool ResultIsPartial, string? Error); public sealed record WorkProgressDto(long Done, long? Total, string? Note); diff --git a/src/Anichron.API/Endpoints/WorkItemDtoMapper.cs b/src/Anichron.API/Endpoints/WorkItemDtoMapper.cs index e81a823..217a51e 100644 --- a/src/Anichron.API/Endpoints/WorkItemDtoMapper.cs +++ b/src/Anichron.API/Endpoints/WorkItemDtoMapper.cs @@ -21,10 +21,12 @@ internal static WorkItemDto ToDto(WorkItemView view) view.StartedAt is { } started ? Format(started) : null, view.FinishedAt is { } finished ? Format(finished) : null, Format(view.UpdatedAt), + view.CancelRequestedAt is { } requested ? Format(requested) : null, new WorkProgressDto(view.Progress.Done, view.Progress.Total, view.Progress.Note), - view is { Status: WorkItemStatus.Succeeded, Result: { } result } + view is { Status: WorkItemStatus.Succeeded or WorkItemStatus.Cancelled, Result: { } result } ? JsonSerializer.Deserialize(result) : null, + view.Status is WorkItemStatus.Cancelled, view.Status is WorkItemStatus.Failed or WorkItemStatus.Cancelled ? view.Error : null); internal static string Format(Instant instant) => InstantPattern.ExtendedIso.Format(instant); diff --git a/src/Anichron.API/Endpoints/WorkItemEndpoints.cs b/src/Anichron.API/Endpoints/WorkItemEndpoints.cs index fb63c13..f566b5a 100644 --- a/src/Anichron.API/Endpoints/WorkItemEndpoints.cs +++ b/src/Anichron.API/Endpoints/WorkItemEndpoints.cs @@ -18,6 +18,7 @@ public static IEndpointRouteBuilder MapWorkItemEndpoints(this IEndpointRouteBuil group.MapGet(string.Empty, ListAsync); group.MapGet(ApiPaths.WorkItems.Events, StreamEvents); group.MapGet(ApiPaths.WorkItems.ById, GetAsync); + group.MapPost(ApiPaths.WorkItems.Cancel, CancelAsync); return app; } @@ -50,6 +51,34 @@ internal static async Task, NotFound>> GetAsync( ? TypedResults.Ok(WorkItemDtoMapper.ToDto(item)) : TypedResults.NotFound(); + /// + /// Cancels a queued item at once (200), or asks the running item's Worker to cancel it at its + /// next heartbeat (202). A finished item is a 409 showing the item as it is. + /// + /// The item to cancel. + /// Makes the status-guarded cancel write. + /// Reads the item after the cancel. + /// Cancels the request. + /// The item after the cancel, or 404 when there is none. + internal static async Task, Accepted, Conflict, NotFound>> CancelAsync( + Guid workItemId, + [FromServices] IWorkItemCanceller canceller, + [FromServices] IWorkItemReader reader, + CancellationToken ct) + { + var outcome = await canceller.CancelAsync(workItemId, ct); + if (outcome == CancelOutcome.NotFound || await reader.FindAsync(workItemId, ct) is not { } item) + return TypedResults.NotFound(); + + var dto = WorkItemDtoMapper.ToDto(item); + return outcome switch + { + CancelOutcome.Cancelled => TypedResults.Ok(dto), + CancelOutcome.Requested => TypedResults.Accepted(ApiPaths.WorkItems.ItemLocation(workItemId), dto), + _ => TypedResults.Conflict(dto), + }; + } + internal static ServerSentEventsResult StreamEvents( [FromHeader(Name = "Last-Event-ID")] string? lastEventId, [FromServices] WorkItemEvents events, diff --git a/src/Anichron.API/Infrastructure/ApiPaths.cs b/src/Anichron.API/Infrastructure/ApiPaths.cs index 76a18f2..f023c1b 100644 --- a/src/Anichron.API/Infrastructure/ApiPaths.cs +++ b/src/Anichron.API/Infrastructure/ApiPaths.cs @@ -56,6 +56,10 @@ internal static class WorkItems internal const string Group = "admin/work-items"; internal const string ById = "{workItemId:guid}"; internal const string Events = "events"; + internal const string Cancel = $"{ById}/cancel"; + + internal static string ItemLocation(Guid workItemId) + => $"{Base}/{Group}/{workItemId}"; } internal static class Flashbacks diff --git a/src/Anichron.API/Infrastructure/ServiceCollectionExtensions.cs b/src/Anichron.API/Infrastructure/ServiceCollectionExtensions.cs index 11554f6..82956a8 100644 --- a/src/Anichron.API/Infrastructure/ServiceCollectionExtensions.cs +++ b/src/Anichron.API/Infrastructure/ServiceCollectionExtensions.cs @@ -85,6 +85,7 @@ public IServiceCollection AddWorkItemServices() .ValidateDataAnnotations() .ValidateOnStart(); services.AddScoped(); + services.AddScoped(); services.AddScoped(); return services; } diff --git a/src/Anichron.Architecture.Tests.Unit/ArchitectureRules.cs b/src/Anichron.Architecture.Tests.Unit/ArchitectureRules.cs index d42018d..a4ed0ea 100644 --- a/src/Anichron.Architecture.Tests.Unit/ArchitectureRules.cs +++ b/src/Anichron.Architecture.Tests.Unit/ArchitectureRules.cs @@ -28,11 +28,14 @@ internal static class ArchitectureRules public const string DomainUsesNodaTime = "domain-uses-nodatime"; public const string OnlyWorkQueueInsertsWorkItems = "only-work-queue-inserts-work-items"; public const string WorkHandlersLiveInWorker = "work-handlers-live-in-worker"; + public const string ApiWritesWorkItemsOnlyByCancel = "api-writes-work-items-only-by-cancel"; private const string Domain = "Anichron.Core.Domain"; private const string ApiWiring = "Anichron.API.Infrastructure"; private const string StagingWriter = "Anichron.Worker.Ingestion.Proxy.ProxyStagingWriter"; private const string WorkQueue = "Anichron.Core.Work.EfWorkQueue"; + private const string WorkItemReader = "Anichron.Core.Work.EfWorkItemReader"; + private const string WorkItemCanceller = "Anichron.Core.Work.EfWorkItemCanceller"; public static IReadOnlyList All { get; } = [ @@ -81,6 +84,13 @@ internal static class ArchitectureRules new(WorkHandlersLiveInWorker, Types().That().ImplementInterface(typeof(IWorkHandler<>)) .Should().ResideInAssembly(WorkerAssembly)), + new(ApiWritesWorkItemsOnlyByCancel, + Types().That().ResideInAssembly(ApiAssembly, CoreAssembly, InfrastructureAssembly) + .And().DoNotHaveFullName(WorkQueue) + .And().DoNotHaveFullName(WorkItemReader) + .And().DoNotHaveFullName(WorkItemCanceller) + .Should().NotCallAny(MethodMembers().That().AreDeclaredIn(typeof(AnichronDbContext)) + .And().HaveNameMatching(@"^get_WorkItems\("))), ]; public static ArchitectureRule Named(string name) => All.Single(rule => rule.Name == name); diff --git a/src/Anichron.Core.Tests.Unit/Work/WorkItemReaderTests.cs b/src/Anichron.Core.Tests.Unit/Work/WorkItemReaderTests.cs index 531be3b..def56af 100644 --- a/src/Anichron.Core.Tests.Unit/Work/WorkItemReaderTests.cs +++ b/src/Anichron.Core.Tests.Unit/Work/WorkItemReaderTests.cs @@ -168,13 +168,15 @@ await SeedAsync(new WorkItem ProgressNote = "7 of 8", Result = /*lang=json,strict*/ """{"found": 8}""", Error = "an earlier error", + CancelRequestedAt = At(5), }); var view = await Reader().FindAsync(First, Ct); view.Should().Be(new WorkItemView( First, WorkKind.Crawl, configId, adminId, At(1), WorkItemStatus.Succeeded, 2, 1, "all@host", - At(2), At(9), At(9), new WorkProgress(7, 8, "7 of 8"), /*lang=json,strict*/ """{"found": 8}""", "an earlier error")); + At(2), At(9), At(9), new WorkProgress(7, 8, "7 of 8"), /*lang=json,strict*/ """{"found": 8}""", "an earlier error", + At(5))); } [Fact] diff --git a/src/Anichron.Core/Data/AnichronDbContext.cs b/src/Anichron.Core/Data/AnichronDbContext.cs index c30a815..49de1f3 100644 --- a/src/Anichron.Core/Data/AnichronDbContext.cs +++ b/src/Anichron.Core/Data/AnichronDbContext.cs @@ -316,8 +316,9 @@ protected override void OnModelCreating(ModelBuilder modelBuilder) { entity.ToTable(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." + + "Only IWorkQueue inserts rows; the API's only other write is an admin's cancel, and " + + "every other state change belongs to the Worker's executor. Delivery is at least " + + "once, so handlers start from scratch and converge." )); entity.Property(w => w.Kind).HasConversion().HasComment( "WorkKind, persisted as text. Picks the payload type and the handler through the " @@ -348,11 +349,16 @@ protected override void OnModelCreating(ModelBuilder modelBuilder) entity.Property(w => w.UpdatedAt).HasComment( "Moves on every insert, claim and transition, and on a heartbeat only when the progress " + "snapshot changed, so a reader can ask what changed since a point in time."); + entity.Property(w => w.CancelRequestedAt).HasComment( + "When an admin asked to cancel the item. A Queued item is cancelled at once; a Running " + + "one keeps running until the Worker's next heartbeat reads this and cancels its handler. " + + "A requested item is never queued again: release and lease recovery end it Cancelled."); entity.Property(w => w.ProgressNote).HasComment( "The latest progress snapshot (done, optional total, note), overwritten by each " + "heartbeat; no history is kept."); entity.Property(w => w.Result).HasColumnType("jsonb").HasComment( - "The handler's results summary, written when the item succeeds."); + "The handler's results summary, written when the item succeeds, or when it is " + + "cancelled after its handler returned one; a Cancelled item's results are partial."); entity.HasIndex(w => w.RequestedAt) .HasFilter(@"""Status"" = 'Queued'") diff --git a/src/Anichron.Core/Domain/WorkItem.cs b/src/Anichron.Core/Domain/WorkItem.cs index 0b86852..ed35f88 100644 --- a/src/Anichron.Core/Domain/WorkItem.cs +++ b/src/Anichron.Core/Domain/WorkItem.cs @@ -18,6 +18,7 @@ public class WorkItem public Instant? LeaseUntil { get; set; } public Instant? FinishedAt { get; set; } public Instant UpdatedAt { get; set; } + public Instant? CancelRequestedAt { get; set; } public long ProgressDone { get; set; } public long? ProgressTotal { get; set; } diff --git a/src/Anichron.Core/Migrations/20261020234633_CancelWorkItems.Designer.cs b/src/Anichron.Core/Migrations/20261020234633_CancelWorkItems.Designer.cs new file mode 100644 index 0000000..a9df0b1 --- /dev/null +++ b/src/Anichron.Core/Migrations/20261020234633_CancelWorkItems.Designer.cs @@ -0,0 +1,697 @@ +// +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("20261020234633_CancelWorkItems")] + partial class CancelWorkItems + { + /// + 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() + .HasDatabaseName("IX_Interactions_UserId_AssetId"); + + 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("CancelRequestedAt") + .HasColumnType("timestamp with time zone") + .HasComment("When an admin asked to cancel the item. A Queued item is cancelled at once; a Running one keeps running until the Worker's next heartbeat reads this and cancels its handler. A requested item is never queued again: release and lease recovery end it Cancelled."); + + 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, or when it is cancelled after its handler returned one; a Cancelled item's results are partial."); + + 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 and transition, and on a heartbeat only when the progress snapshot changed, 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; the API's only other write is an admin's cancel, and every other 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/20261020234633_CancelWorkItems.cs b/src/Anichron.Core/Migrations/20261020234633_CancelWorkItems.cs new file mode 100644 index 0000000..8195b65 --- /dev/null +++ b/src/Anichron.Core/Migrations/20261020234633_CancelWorkItems.cs @@ -0,0 +1,62 @@ +using Microsoft.EntityFrameworkCore.Migrations; +using NodaTime; + +#nullable disable + +namespace Anichron.Core.Migrations +{ + /// + public partial class CancelWorkItems : Migration + { + /// + protected override void Up(MigrationBuilder migrationBuilder) + { + migrationBuilder.AlterTable( + name: "WorkItems", + comment: "Durable work for the Worker: periodic maintenance today, admin requests later. Only IWorkQueue inserts rows; the API's only other write is an admin's cancel, and every other state change belongs to the Worker's executor. Delivery is at least once, so handlers start from scratch and converge.", + oldComment: "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."); + + migrationBuilder.AlterColumn( + name: "Result", + table: "WorkItems", + type: "jsonb", + nullable: true, + comment: "The handler's results summary, written when the item succeeds, or when it is cancelled after its handler returned one; a Cancelled item's results are partial.", + oldClrType: typeof(string), + oldType: "jsonb", + oldNullable: true, + oldComment: "The handler's results summary, written when the item succeeds."); + + migrationBuilder.AddColumn( + name: "CancelRequestedAt", + table: "WorkItems", + type: "timestamp with time zone", + nullable: true, + comment: "When an admin asked to cancel the item. A Queued item is cancelled at once; a Running one keeps running until the Worker's next heartbeat reads this and cancels its handler. A requested item is never queued again: release and lease recovery end it Cancelled."); + } + + /// + protected override void Down(MigrationBuilder migrationBuilder) + { + migrationBuilder.DropColumn( + name: "CancelRequestedAt", + table: "WorkItems"); + + migrationBuilder.AlterTable( + name: "WorkItems", + comment: "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.", + oldComment: "Durable work for the Worker: periodic maintenance today, admin requests later. Only IWorkQueue inserts rows; the API's only other write is an admin's cancel, and every other state change belongs to the Worker's executor. Delivery is at least once, so handlers start from scratch and converge."); + + migrationBuilder.AlterColumn( + name: "Result", + table: "WorkItems", + type: "jsonb", + nullable: true, + comment: "The handler's results summary, written when the item succeeds.", + oldClrType: typeof(string), + oldType: "jsonb", + oldNullable: true, + oldComment: "The handler's results summary, written when the item succeeds, or when it is cancelled after its handler returned one; a Cancelled item's results are partial."); + } + } +} diff --git a/src/Anichron.Core/Migrations/AnichronDbContextModelSnapshot.cs b/src/Anichron.Core/Migrations/AnichronDbContextModelSnapshot.cs index d3c2dd7..5c6a02e 100644 --- a/src/Anichron.Core/Migrations/AnichronDbContextModelSnapshot.cs +++ b/src/Anichron.Core/Migrations/AnichronDbContextModelSnapshot.cs @@ -432,6 +432,10 @@ protected override void BuildModel(ModelBuilder modelBuilder) .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("CancelRequestedAt") + .HasColumnType("timestamp with time zone") + .HasComment("When an admin asked to cancel the item. A Queued item is cancelled at once; a Running one keeps running until the Worker's next heartbeat reads this and cancels its handler. A requested item is never queued again: release and lease recovery end it Cancelled."); + b.Property("ClaimedBy") .HasColumnType("text") .HasComment("The claiming Worker for display: its WORKER__USER or 'all', then its host name."); @@ -479,7 +483,7 @@ protected override void BuildModel(ModelBuilder modelBuilder) b.Property("Result") .HasColumnType("jsonb") - .HasComment("The handler's results summary, written when the item succeeds."); + .HasComment("The handler's results summary, written when the item succeeds, or when it is cancelled after its handler returned one; a Cancelled item's results are partial."); b.Property("Stalls") .HasColumnType("integer") @@ -524,7 +528,7 @@ protected override void BuildModel(ModelBuilder modelBuilder) 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."); + t.HasComment("Durable work for the Worker: periodic maintenance today, admin requests later. Only IWorkQueue inserts rows; the API's only other write is an admin's cancel, and every other state change belongs to the Worker's executor. Delivery is at least once, so handlers start from scratch and converge."); }); }); diff --git a/src/Anichron.Core/Work/WorkItemCanceller.cs b/src/Anichron.Core/Work/WorkItemCanceller.cs new file mode 100644 index 0000000..cc9241d --- /dev/null +++ b/src/Anichron.Core/Work/WorkItemCanceller.cs @@ -0,0 +1,46 @@ +using Anichron.Core.Data; +using Anichron.Core.Domain; +using Microsoft.EntityFrameworkCore; + +namespace Anichron.Core.Work; + +public enum CancelOutcome +{ + /// The item was queued and is now cancelled; it never starts. + Cancelled = 1, + + /// The item is running; its Worker cancels the handler at the next heartbeat. + Requested = 2, + + Finished = 3, + NotFound = 4, +} + +public interface IWorkItemCanceller +{ + Task CancelAsync(Guid id, CancellationToken ct); +} + +public sealed class EfWorkItemCanceller(AnichronDbContext db, IClock clock) : IWorkItemCanceller +{ + // Stryker disable all + public async Task CancelAsync(Guid id, CancellationToken ct) + { + var now = clock.GetCurrentInstant(); + var statuses = await db.Database.SqlQuery($""" + UPDATE "WorkItems" + SET "Status" = CASE WHEN "Status" = 'Queued' THEN 'Cancelled' ELSE "Status" END, + "FinishedAt" = CASE WHEN "Status" = 'Queued' THEN {now} ELSE "FinishedAt" END, + "CancelRequestedAt" = COALESCE("CancelRequestedAt", {now}), + "UpdatedAt" = CASE WHEN "CancelRequestedAt" IS NULL THEN {now} ELSE "UpdatedAt" END + WHERE "Id" = {id} AND "Status" IN ('Queued', 'Running') + RETURNING "Status" AS "Value" + """).ToListAsync(ct); + return statuses switch + { + [nameof(WorkItemStatus.Cancelled)] => CancelOutcome.Cancelled, + [_] => CancelOutcome.Requested, + _ => await db.WorkItems.AnyAsync(w => w.Id == id, ct) ? CancelOutcome.Finished : CancelOutcome.NotFound, + }; + } +} diff --git a/src/Anichron.Core/Work/WorkItemReader.cs b/src/Anichron.Core/Work/WorkItemReader.cs index 687efad..803bc36 100644 --- a/src/Anichron.Core/Work/WorkItemReader.cs +++ b/src/Anichron.Core/Work/WorkItemReader.cs @@ -20,7 +20,8 @@ public sealed record WorkItemView( Instant UpdatedAt, WorkProgress Progress, string? Result, - string? Error); + string? Error, + Instant? CancelRequestedAt = null); public sealed record WorkItemQuery(WorkItemStatus? Status, WorkKind? Kind, int Skip, int Take); @@ -88,5 +89,6 @@ private static Expression> ToView => w => new WorkItemView( w.Id, w.Kind, w.StorageConfigId, w.RequestedBy, w.RequestedAt, w.Status, w.Attempt, w.Stalls, w.ClaimedBy, w.StartedAt, w.FinishedAt, w.UpdatedAt, - new WorkProgress(w.ProgressDone, w.ProgressTotal, w.ProgressNote), w.Result, w.Error); + new WorkProgress(w.ProgressDone, w.ProgressTotal, w.ProgressNote), w.Result, w.Error, + w.CancelRequestedAt); } diff --git a/src/Anichron.Tests.Integration/Work/WorkDatabase.cs b/src/Anichron.Tests.Integration/Work/WorkDatabase.cs index bcf5317..96df39e 100644 --- a/src/Anichron.Tests.Integration/Work/WorkDatabase.cs +++ b/src/Anichron.Tests.Integration/Work/WorkDatabase.cs @@ -46,6 +46,12 @@ public async Task EnqueueAsync(TPayload payload, CancellationTok return await new EfWorkQueue(db, Clock, new TimeOrderedGuidFactory()).EnqueueAsync(payload, null, ct); } + public async Task CancelAsync(Guid id, CancellationToken ct) + { + await using var db = CreateDbContext(); + return await new EfWorkItemCanceller(db, Clock).CancelAsync(id, ct); + } + public async Task ItemAsync(Guid id, CancellationToken ct) { await using var db = CreateDbContext(); diff --git a/src/Anichron.Tests.Integration/Work/WorkItemCancellerTests.cs b/src/Anichron.Tests.Integration/Work/WorkItemCancellerTests.cs new file mode 100644 index 0000000..c2e507b --- /dev/null +++ b/src/Anichron.Tests.Integration/Work/WorkItemCancellerTests.cs @@ -0,0 +1,122 @@ +using Anichron.Core.Domain; +using Anichron.Core.Work; +using Anichron.Worker.Work; + +namespace Anichron.Tests.Integration.Work; + +public sealed class WorkItemCancellerTests(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); + + [Fact] + public async Task CancelAsync_QueuedItem_CancelsItAtOnceSoItNeverStartsAsync() + { + var work = await WorkDatabase.CreateAsync(server, Ct); + var id = await work.EnqueueAsync(new UnkeyedPayload(1), Ct); + work.Clock.Now = At(5); + + var outcome = await work.CancelAsync(id, Ct); + + outcome.Should().Be(CancelOutcome.Cancelled); + var item = await work.ItemAsync(id, Ct); + (item.Status, item.Error).Should().Be((WorkItemStatus.Cancelled, null)); + (item.CancelRequestedAt, item.FinishedAt, item.UpdatedAt).Should().Be((At(5), At(5), At(5))); + (await work.Store.ClaimAsync("worker", null, At(6), Lease, Ct)).Should().BeNull(); + } + + [Fact] + public async Task CancelAsync_RunningItem_RecordsTheRequestAndLeavesItRunningAsync() + { + var work = await WorkDatabase.CreateAsync(server, Ct); + var id = await work.EnqueueAsync(new UnkeyedPayload(1), Ct); + await work.Store.ClaimAsync("worker", null, At(1), Lease, Ct); + work.Clock.Now = At(5); + + var outcome = await work.CancelAsync(id, Ct); + + outcome.Should().Be(CancelOutcome.Requested); + var item = await work.ItemAsync(id, Ct); + (item.Status, item.FinishedAt).Should().Be((WorkItemStatus.Running, null)); + (item.CancelRequestedAt, item.UpdatedAt).Should().Be((At(5), At(5))); + } + + [Fact] + public async Task CancelAsync_RunningItemAlreadyRequested_KeepsTheFirstRequestAsync() + { + var work = await WorkDatabase.CreateAsync(server, Ct); + var id = await work.EnqueueAsync(new UnkeyedPayload(1), Ct); + await work.Store.ClaimAsync("worker", null, At(1), Lease, Ct); + work.Clock.Now = At(5); + await work.CancelAsync(id, Ct); + work.Clock.Now = At(8); + + var outcome = await work.CancelAsync(id, Ct); + + outcome.Should().Be(CancelOutcome.Requested); + (await work.ItemAsync(id, Ct)).CancelRequestedAt.Should().Be(At(5)); + } + + [Theory] + [InlineData(WorkItemStatus.Succeeded)] + [InlineData(WorkItemStatus.Failed)] + [InlineData(WorkItemStatus.Cancelled)] + public async Task CancelAsync_FinishedItem_ReportsItFinishedAndChangesNothingAsync(WorkItemStatus finished) + { + var work = await WorkDatabase.CreateAsync(server, Ct); + var id = await work.EnqueueAsync(new UnkeyedPayload(1), Ct); + var claimed = (await work.Store.ClaimAsync("worker", null, At(1), Lease, Ct))!; + if (finished == WorkItemStatus.Succeeded) + await work.Store.SucceedAsync(claimed.Attempt, WorkProgress.None, null, At(2), Ct); + else if (finished == WorkItemStatus.Failed) + await work.Store.FailAsync(claimed.Attempt, WorkProgress.None, "boom", At(2), Ct); + else + await work.Store.CancelAsync(claimed.Attempt, WorkProgress.None, null, At(2), Ct); + work.Clock.Now = At(5); + + var outcome = await work.CancelAsync(id, Ct); + + outcome.Should().Be(CancelOutcome.Finished); + var item = await work.ItemAsync(id, Ct); + (item.Status, item.CancelRequestedAt, item.UpdatedAt).Should().Be((finished, null, At(2))); + } + + [Fact] + public async Task CancelAsync_UnknownId_ReportsNotFoundAsync() + { + var work = await WorkDatabase.CreateAsync(server, Ct); + await work.EnqueueAsync(new UnkeyedPayload(1), Ct); + + (await work.CancelAsync(Guid.NewGuid(), Ct)).Should().Be(CancelOutcome.NotFound); + } + + [Fact] + public async Task CancelAsync_RacingAClaim_NeverLosesTheCancelAsync() + { + var work = await WorkDatabase.CreateAsync(server, Ct); + + for (var round = 0; round < 40; round++) + { + var id = await work.EnqueueAsync(new UnkeyedPayload(round), Ct); + var claim = Task.Run(() => work.Store.ClaimAsync("worker", null, At(round), Lease, Ct), Ct); + var cancel = Task.Run(() => work.CancelAsync(id, Ct), Ct); + var (claimed, outcome) = (await claim, await cancel); + + if (claimed is null) + { + outcome.Should().Be(CancelOutcome.Cancelled); + (await work.ItemAsync(id, Ct)).Status.Should().Be(WorkItemStatus.Cancelled); + } + else + { + outcome.Should().Be(CancelOutcome.Requested); + (await work.Store.HeartbeatAsync(claimed.Attempt, WorkProgress.None, At(round), Lease, Ct)) + .Should().Be(HeartbeatOutcome.CancelRequested); + await work.Store.CancelAsync(claimed.Attempt, WorkProgress.None, null, At(round), Ct); + } + } + } +} diff --git a/src/Anichron.Tests.Integration/Work/WorkItemStoreTests.cs b/src/Anichron.Tests.Integration/Work/WorkItemStoreTests.cs index 7c433b2..9de6b28 100644 --- a/src/Anichron.Tests.Integration/Work/WorkItemStoreTests.cs +++ b/src/Anichron.Tests.Integration/Work/WorkItemStoreTests.cs @@ -84,7 +84,7 @@ public async Task HeartbeatAsync_HoldingAttempt_ExtendsTheLeaseAndStoresTheProgr var held = await work.Store.HeartbeatAsync(claimed.Attempt, Halfway, At(30), Lease, Ct); - held.Should().BeTrue(); + held.Should().Be(HeartbeatOutcome.Held); var item = await work.ItemAsync(claimed.Attempt.Id, Ct); item.LeaseUntil.Should().Be(At(90)); (item.ProgressDone, item.ProgressTotal, item.ProgressNote).Should().Be((5L, 10L, "halfway")); @@ -101,7 +101,7 @@ public async Task HeartbeatAsync_SameProgress_ExtendsTheLeaseWithoutMovingUpdate var held = await work.Store.HeartbeatAsync(claimed.Attempt, new WorkProgress(5, 10, "halfway"), At(32), Lease, Ct); - held.Should().BeTrue(); + held.Should().Be(HeartbeatOutcome.Held); var item = await work.ItemAsync(claimed.Attempt.Id, Ct); item.LeaseUntil.Should().Be(At(92)); item.UpdatedAt.Should().Be(At(30)); @@ -154,8 +154,9 @@ public async Task AfterAReclaim_TheOldAttemptsHeartbeatAndCompletionChangeNothin await work.Store.SucceedAsync(zombie.Attempt, Halfway, /*lang=json,strict*/ """{"done": true}""", At(64), Ct); await work.Store.FailAsync(zombie.Attempt, Halfway, "boom", At(64), Ct); await work.Store.ReleaseAsync(zombie.Attempt, At(64), Ct); + await work.Store.CancelAsync(zombie.Attempt, Halfway, /*lang=json,strict*/ """{"done": false}""", At(64), Ct); - held.Should().BeFalse(); + held.Should().Be(HeartbeatOutcome.Lost); current.Attempt.Should().Be(new WorkAttempt(zombie.Attempt.Id, 2)); var item = await work.ItemAsync(current.Attempt.Id, Ct); item.Status.Should().Be(WorkItemStatus.Running); @@ -339,6 +340,90 @@ public async Task FailAsync_StoresTheErrorAndTheLastProgressButNoResultAsync() (item.FinishedAt, item.UpdatedAt).Should().Be((At(9), At(9))); } + [Fact] + public async Task HeartbeatAsync_CancelRequested_ReportsItAndStillExtendsTheLeaseAsync() + { + var work = await WorkDatabase.CreateAsync(server, Ct); + var id = await work.EnqueueAsync(new UnkeyedPayload(1), Ct); + var claimed = (await work.Store.ClaimAsync("worker", null, At(0), Lease, Ct))!; + work.Clock.Now = At(10); + await work.CancelAsync(id, Ct); + + var heartbeat = await work.Store.HeartbeatAsync(claimed.Attempt, Halfway, At(30), Lease, Ct); + + heartbeat.Should().Be(HeartbeatOutcome.CancelRequested); + var item = await work.ItemAsync(id, Ct); + (item.Status, item.LeaseUntil).Should().Be((WorkItemStatus.Running, At(90))); + item.ProgressNote.Should().Be("halfway"); + } + + [Fact] + public async Task CancelAsync_StoresTheLastProgressAndTheResultAsPartialAsync() + { + var work = await WorkDatabase.CreateAsync(server, Ct); + var id = await work.EnqueueAsync(new UnkeyedPayload(1), Ct); + var claimed = (await work.Store.ClaimAsync("worker", null, At(0), Lease, Ct))!; + work.Clock.Now = At(5); + await work.CancelAsync(id, Ct); + + await work.Store.CancelAsync(claimed.Attempt, Halfway, /*lang=json,strict*/ """{"deleted": 2}""", At(9), Ct); + + var item = await work.ItemAsync(id, Ct); + (item.Status, item.Error).Should().Be((WorkItemStatus.Cancelled, null)); + item.Result.Should().Be(/*lang=json,strict*/ """{"deleted": 2}"""); + (item.ProgressDone, item.ProgressTotal, item.ProgressNote).Should().Be((5L, 10L, "halfway")); + (item.FinishedAt, item.UpdatedAt).Should().Be((At(9), At(9))); + } + + [Fact] + public async Task ReleaseAsync_CancelRequested_CancelsTheItemInsteadOfQueuingItAsync() + { + var work = await WorkDatabase.CreateAsync(server, Ct); + var id = await work.EnqueueAsync(new UnkeyedPayload(1), Ct); + var claimed = (await work.Store.ClaimAsync("worker", null, At(0), Lease, Ct))!; + work.Clock.Now = At(3); + await work.CancelAsync(id, Ct); + + await work.Store.ReleaseAsync(claimed.Attempt, At(5), Ct); + + var item = await work.ItemAsync(id, Ct); + (item.Status, item.Error, item.Stalls).Should().Be((WorkItemStatus.Cancelled, null, 0)); + (item.FinishedAt, item.UpdatedAt).Should().Be((At(5), At(5))); + (await work.Store.ClaimAsync("worker", null, At(6), Lease, Ct)).Should().BeNull(); + } + + [Fact] + public async Task ReleaseAsync_CancelRequestedAndSameKeyQueued_CancelsWithoutCallingItSupersededAsync() + { + var work = await WorkDatabase.CreateAsync(server, Ct); + var id = await work.EnqueueAsync(new KeyedPayload("sweep"), Ct); + var claimed = (await work.Store.ClaimAsync("worker", null, At(0), Lease, Ct))!; + await work.CancelAsync(id, Ct); + var waiting = await work.EnqueueAsync(new KeyedPayload("sweep"), Ct); + + await work.Store.ReleaseAsync(claimed.Attempt, At(5), Ct); + + (await work.ItemAsync(id, Ct)).Should().BeEquivalentTo(new { Status = WorkItemStatus.Cancelled, Error = (string?)null }); + (await work.ItemAsync(waiting, Ct)).Status.Should().Be(WorkItemStatus.Queued); + } + + [Theory] + [InlineData(3)] + [InlineData(1)] + public async Task RecoverExpiredLeasesAsync_CancelRequested_CancelsTheItemEvenAtTheStallLimitAsync(int maxStalls) + { + var work = await WorkDatabase.CreateAsync(server, Ct); + var id = await work.EnqueueAsync(new UnkeyedPayload(1), Ct); + await work.Store.ClaimAsync("worker", null, At(0), Lease, Ct); + await work.CancelAsync(id, Ct); + + await work.Store.RecoverExpiredLeasesAsync(At(61), maxStalls, Ct); + + var item = await work.ItemAsync(id, Ct); + (item.Status, item.Error, item.Stalls).Should().Be((WorkItemStatus.Cancelled, null, 1)); + item.FinishedAt.Should().Be(At(61)); + } + [Fact] public async Task DeleteFinishedBeforeAsync_DeletesOnlyItemsThatFinishedBeforeTheCutoffAsync() { diff --git a/src/Anichron.Tests.Integration/Work/WorkScopeTests.cs b/src/Anichron.Tests.Integration/Work/WorkScopeTests.cs index 9bb2d01..4ac2986 100644 --- a/src/Anichron.Tests.Integration/Work/WorkScopeTests.cs +++ b/src/Anichron.Tests.Integration/Work/WorkScopeTests.cs @@ -1,5 +1,6 @@ using Anichron.Core.Domain; using Anichron.Core.Work; +using Anichron.Worker.Work; using Microsoft.EntityFrameworkCore; namespace Anichron.Tests.Integration.Work; @@ -104,6 +105,6 @@ public async Task DeletingAConfig_DeletesItsItemsAndFencesTheRunningOneAsync() 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(); + (await library.Work.Store.HeartbeatAsync(running.Attempt, WorkProgress.None, At(5), Lease, Ct)).Should().Be(HeartbeatOutcome.Lost); } } diff --git a/src/Anichron.Worker.Tests.Unit/Work/WorkExecutorTests.cs b/src/Anichron.Worker.Tests.Unit/Work/WorkExecutorTests.cs index 5570d2d..341269d 100644 --- a/src/Anichron.Worker.Tests.Unit/Work/WorkExecutorTests.cs +++ b/src/Anichron.Worker.Tests.Unit/Work/WorkExecutorTests.cs @@ -58,7 +58,7 @@ public WorkExecutorTests() .Returns(_ => { timeline.Record("heartbeat"); - return true; + return HeartbeatOutcome.Held; }); store.When(s => s.SucceedAsync( Arg.Any(), Arg.Any(), Arg.Any(), Arg.Any(), Arg.Any())) @@ -66,6 +66,9 @@ public WorkExecutorTests() store.When(s => s.FailAsync( Arg.Any(), Arg.Any(), Arg.Any(), Arg.Any(), Arg.Any())) .Do(_ => timeline.Record("fail")); + store.When(s => s.CancelAsync( + Arg.Any(), Arg.Any(), Arg.Any(), Arg.Any(), Arg.Any())) + .Do(_ => timeline.Record("cancel")); store.When(s => s.ReleaseAsync(Arg.Any(), Arg.Any(), Arg.Any())) .Do(_ => timeline.Record("release")); handler.RunAsync(Arg.Any(), Arg.Any(), Arg.Any()) @@ -229,7 +232,7 @@ public async Task ExecuteAsync_HeartbeatFails_LogsAndKeepsHeartbeatingAsync() .Returns(_ => { timeline.Record("heartbeat"); - return ++beats == 1 ? throw new InvalidOperationException("db down") : true; + return ++beats == 1 ? throw new InvalidOperationException("db down") : HeartbeatOutcome.Held; }); await StartRunningAsync(); @@ -250,7 +253,7 @@ public async Task ExecuteAsync_HeartbeatFenced_CancelsTheHandlerAndWritesNothing .Returns(_ => { timeline.Record("heartbeat"); - return false; + return HeartbeatOutcome.Lost; }); await StartRunningAsync(); @@ -271,7 +274,7 @@ public async Task ExecuteAsync_FencedHandlerFinishesAnyway_WritesNothingAsync() { timeline.Record("heartbeat"); fenced = runs[0].TrySetResult("done anyway"); - return false; + return HeartbeatOutcome.Lost; }); await StartRunningAsync(); @@ -282,6 +285,125 @@ await timeline.ExpectAsync( fenced.Should().BeTrue(); } + // ========================================================================== + // Cancel requests + // ========================================================================== + + private void HeartbeatsReportACancelRequest() + => store.HeartbeatAsync( + Arg.Any(), Arg.Any(), Arg.Any(), Arg.Any(), Arg.Any()) + .Returns(_ => + { + timeline.Record("heartbeat"); + return HeartbeatOutcome.CancelRequested; + }); + + private void HandlerWindsDownOnlyWhenTold() + => handler.RunAsync(Arg.Any(), Arg.Any(), Arg.Any()) + .Returns(call => + { + handlerProgress = call.Arg(); + handlerToken = call.Arg(); + var run = new TaskCompletionSource(TaskCreationOptions.RunContinuationsAsynchronously); + runs.Add(run); + timeline.Record("run"); + return run.Task; + }); + + private async Task CancelRequestedWhileRunningAsync() + { + HeartbeatsReportACancelRequest(); + HandlerWindsDownOnlyWhenTold(); + await StartRunningAsync(); + timeline.Clock.Advance(Heartbeat); + await timeline.ExpectAsync(new TimerFired(Heartbeat), new Called("heartbeat"), new TimerArmed(Heartbeat)); + } + + [Fact] + public async Task ExecuteAsync_CancelRequested_CancelsTheHandlerAndKeepsHoldingTheLeaseAsync() + { + await CancelRequestedWhileRunningAsync(); + + handlerToken.IsCancellationRequested.Should().BeTrue(); + timeline.Clock.Advance(Heartbeat); + await timeline.ExpectAsync(new TimerFired(Heartbeat), new Called("heartbeat"), new TimerArmed(Heartbeat)); + runs[0].SetCanceled(handlerToken); + await timeline.ExpectAsync(new Called("cancel"), new Called("claim"), new TimerArmed(Poll)); + } + + [Fact] + public async Task ExecuteAsync_CancelRequestedAndHandlerStops_CancelsTheItemWithItsLastProgressAsync() + { + await CancelRequestedWhileRunningAsync(); + handlerProgress!.Report(4, null, "walking"); + + runs[0].SetCanceled(handlerToken); + + await timeline.ExpectAsync(new Called("cancel"), new Called("claim"), new TimerArmed(Poll)); + await store.Received(1).CancelAsync( + Work().Attempt, new WorkProgress(4, null, "walking"), null, Now, CancellationToken.None); + logger.Collector.GetSnapshot().Should().ContainSingle(e => e.Level == LogLevel.Information) + .Which.Message.Should().Be($"Work item {Work().Attempt.Id} was cancelled at an admin's request."); + } + + [Fact] + public async Task ExecuteAsync_CancelRequestedButHandlerReturnsASummary_KeepsItAsThePartialResultAsync() + { + await CancelRequestedWhileRunningAsync(); + + runs[0].SetResult(new { Deleted = 2 }); + + await timeline.ExpectAsync(new Called("cancel"), new Called("claim"), new TimerArmed(Poll)); + await store.Received(1).CancelAsync( + Work().Attempt, WorkProgress.None, /*lang=json,strict*/ """{"deleted":2}""", Now, CancellationToken.None); + } + + [Fact] + public async Task ExecuteAsync_CancelRequestedButHandlerFails_FailsTheItemAsync() + { + await CancelRequestedWhileRunningAsync(); + + runs[0].SetException(new InvalidOperationException("boom")); + + await timeline.ExpectAsync(new Called("fail"), new Called("claim"), new TimerArmed(Poll)); + } + + [Fact] + public async Task StopAsync_AfterACancelRequest_CancelsTheItemInsteadOfReleasingItAsync() + { + await CancelRequestedWhileRunningAsync(); + + var stopping = executor!.StopAsync(Ct); + runs[0].SetCanceled(handlerToken); + await stopping; + + await timeline.ExpectAsync(new Called("cancel"), new Called("claim"), new Finished(TaskStatus.Canceled)); + await store.DidNotReceive().ReleaseAsync(Arg.Any(), Arg.Any(), Arg.Any()); + } + + [Fact] + public async Task ExecuteAsync_CancelRequestedThenFenced_WritesNothingAsync() + { + var beats = 0; + store.HeartbeatAsync( + Arg.Any(), Arg.Any(), Arg.Any(), Arg.Any(), Arg.Any()) + .Returns(_ => + { + timeline.Record("heartbeat"); + return ++beats == 1 ? HeartbeatOutcome.CancelRequested : HeartbeatOutcome.Lost; + }); + HandlerWindsDownOnlyWhenTold(); + await StartRunningAsync(); + timeline.Clock.Advance(Heartbeat); + await timeline.ExpectAsync(new TimerFired(Heartbeat), new Called("heartbeat"), new TimerArmed(Heartbeat)); + + timeline.Clock.Advance(Heartbeat); + await timeline.ExpectAsync(new TimerFired(Heartbeat), new Called("heartbeat")); + runs[0].SetCanceled(handlerToken); + + await timeline.ExpectAsync(new Called("claim"), new TimerArmed(Poll)); + } + // ========================================================================== // Outcomes // ========================================================================== diff --git a/src/Anichron.Worker/Work/EfWorkItemStore.cs b/src/Anichron.Worker/Work/EfWorkItemStore.cs index 492e6cb..0b88aeb 100644 --- a/src/Anichron.Worker/Work/EfWorkItemStore.cs +++ b/src/Anichron.Worker/Work/EfWorkItemStore.cs @@ -44,7 +44,7 @@ LIMIT 1 } // Stryker disable all - public async Task HeartbeatAsync( + public async Task HeartbeatAsync( WorkAttempt attempt, WorkProgress progress, Instant now, Duration lease, CancellationToken ct) { await using var db = await contexts.CreateDbContextAsync(ct); @@ -58,7 +58,11 @@ public async Task HeartbeatAsync( w => w.ProgressDone == progress.Done && w.ProgressTotal == progress.Total && w.ProgressNote == progress.Note ? w.UpdatedAt : now), ct); - return rows == 1; + if (rows == 0) + return HeartbeatOutcome.Lost; + return await Running(db, attempt).AnyAsync(w => w.CancelRequestedAt != null, ct) + ? HeartbeatOutcome.CancelRequested + : HeartbeatOutcome.Held; } // Stryker disable all @@ -91,6 +95,21 @@ await Running(db, attempt).ExecuteUpdateAsync(s => s .SetProperty(w => w.UpdatedAt, now), ct); } + // Stryker disable all + public async Task CancelAsync( + WorkAttempt attempt, WorkProgress progress, string? result, Instant now, CancellationToken ct) + { + await using var db = await contexts.CreateDbContextAsync(ct); + await Running(db, attempt).ExecuteUpdateAsync(s => s + .SetProperty(w => w.Status, WorkItemStatus.Cancelled) + .SetProperty(w => w.ProgressDone, progress.Done) + .SetProperty(w => w.ProgressTotal, progress.Total) + .SetProperty(w => w.ProgressNote, progress.Note) + .SetProperty(w => w.Result, result) + .SetProperty(w => w.FinishedAt, now) + .SetProperty(w => w.UpdatedAt, now), ct); + } + // Stryker disable all public async Task ReleaseAsync(WorkAttempt attempt, Instant now, CancellationToken ct) { @@ -102,7 +121,7 @@ public async Task ReleaseAsync(WorkAttempt attempt, Instant now, CancellationTok public async Task RecoverExpiredLeasesAsync(Instant now, int maxStalls, CancellationToken ct) { await using var db = await contexts.CreateDbContextAsync(ct); - await Expired(db, now).Where(w => w.Stalls + 1 >= maxStalls).ExecuteUpdateAsync(s => s + await Expired(db, now).Where(w => w.Stalls + 1 >= maxStalls && w.CancelRequestedAt == null).ExecuteUpdateAsync(s => s .SetProperty(w => w.Status, WorkItemStatus.Failed) .SetProperty(w => w.Stalls, w => w.Stalls + 1) .SetProperty(w => w.Error, StalledError) @@ -138,7 +157,7 @@ private static async Task TryReturnToQueueOnceAsync( try { var requeued = await running - .Where(w => !db.WorkItems.Any(q => + .Where(w => w.CancelRequestedAt == null && !db.WorkItems.Any(q => q.Status == WorkItemStatus.Queued && q.DedupKey != null && q.DedupKey == w.DedupKey)) .ExecuteUpdateAsync(s => s .SetProperty(w => w.Status, WorkItemStatus.Queued) @@ -152,7 +171,7 @@ private static async Task TryReturnToQueueOnceAsync( await running.ExecuteUpdateAsync(s => s .SetProperty(w => w.Status, WorkItemStatus.Cancelled) .SetProperty(w => w.Stalls, w => w.Stalls + stall) - .SetProperty(w => w.Error, SupersededError) + .SetProperty(w => w.Error, w => w.CancelRequestedAt == null ? SupersededError : null) .SetProperty(w => w.FinishedAt, now) .SetProperty(w => w.UpdatedAt, now), ct); } diff --git a/src/Anichron.Worker/Work/IWorkItemStore.cs b/src/Anichron.Worker/Work/IWorkItemStore.cs index afc5e8c..e9bef30 100644 --- a/src/Anichron.Worker/Work/IWorkItemStore.cs +++ b/src/Anichron.Worker/Work/IWorkItemStore.cs @@ -8,6 +8,17 @@ public sealed record WorkAttempt(Guid Id, int Attempt); public sealed record ClaimedWork(WorkAttempt Attempt, WorkKind Kind, string Payload); +public enum HeartbeatOutcome +{ + Held = 1, + + /// The attempt still holds the item, and an admin asked to cancel it. + CancelRequested = 2, + + /// The attempt no longer holds the item. + Lost = 3, +} + /// /// Every state change a Worker makes to a work item after it was enqueued. Writes about a running /// item are fenced by its : once the item was recovered and claimed again, @@ -33,14 +44,26 @@ public interface IWorkItemStore /// The heartbeat time. /// How long the lease holds from now. /// Cancels the heartbeat. - /// False when the attempt no longer holds the item. - Task HeartbeatAsync(WorkAttempt attempt, WorkProgress progress, Instant now, Duration lease, CancellationToken ct); + /// Whether the attempt still holds the item, and whether a cancel was requested. + Task HeartbeatAsync(WorkAttempt attempt, WorkProgress progress, Instant now, Duration lease, CancellationToken ct); Task SucceedAsync(WorkAttempt attempt, WorkProgress progress, string? result, Instant now, CancellationToken ct); Task FailAsync(WorkAttempt attempt, WorkProgress progress, string error, Instant now, CancellationToken ct); - /// Hands the item back to the queue without counting a stall. + /// Ends a running item whose cancel was requested, keeping what the attempt produced. + /// The running attempt. + /// The last snapshot. + /// The handler's summary, if it returned one; partial either way. + /// The cancel time. + /// Cancels the write. + /// A task that completes once the item is cancelled. + Task CancelAsync(WorkAttempt attempt, WorkProgress progress, string? result, Instant now, CancellationToken ct); + + /// + /// Hands the item back to the queue without counting a stall, or cancels it when a cancel was + /// requested. + /// /// The running attempt. /// The release time. /// Cancels the release. @@ -49,7 +72,8 @@ public interface IWorkItemStore /// /// Puts running items whose lease expired back in the queue, counting a stall, and fails those - /// reaching as stalled. + /// reaching as stalled. An item whose cancel was requested is + /// cancelled instead. /// /// Leases ending before this have expired. /// The stall that fails an item. diff --git a/src/Anichron.Worker/Work/WorkExecutor.cs b/src/Anichron.Worker/Work/WorkExecutor.cs index 4a17578..14845fc 100644 --- a/src/Anichron.Worker/Work/WorkExecutor.cs +++ b/src/Anichron.Worker/Work/WorkExecutor.cs @@ -64,10 +64,17 @@ private async Task RunAsync(ClaimedWork work, CancellationToken stopping) } await done.CancelAsync(); - if (await heartbeat) + var lastHeartbeat = await heartbeat; + if (lastHeartbeat == HeartbeatOutcome.Lost) { Log.Fenced(logger, work.Attempt.Id); } + else if (lastHeartbeat == HeartbeatOutcome.CancelRequested && failure is null or OperationCanceledException) + { + await store.CancelAsync( + work.Attempt, progress.Snapshot, WorkJson.SerializeResult(result), Now, CancellationToken.None); + Log.Cancelled(logger, work.Attempt.Id); + } else if (failure is null) { await store.SucceedAsync( @@ -87,10 +94,11 @@ await store.FailAsync( } } - private async Task HeartbeatAsync( + private async Task HeartbeatAsync( WorkAttempt attempt, WorkProgressSink progress, CancellationTokenSource running, CancellationToken done) { var interval = TimeSpan.FromSeconds(settings.WorkHeartbeatSeconds); + var held = HeartbeatOutcome.Held; try { while (true) @@ -98,11 +106,12 @@ private async Task HeartbeatAsync( await Task.Delay(interval, timeProvider, done); try { - if (!await store.HeartbeatAsync(attempt, progress.Snapshot, Now, Lease, CancellationToken.None)) - { + var beat = await store.HeartbeatAsync(attempt, progress.Snapshot, Now, Lease, CancellationToken.None); + if (beat != HeartbeatOutcome.Held) await running.CancelAsync(); - return true; - } + if (beat == HeartbeatOutcome.Lost) + return beat; + held = beat; } catch (Exception ex) { @@ -112,7 +121,7 @@ private async Task HeartbeatAsync( } catch (OperationCanceledException) when (done.IsCancellationRequested) { - return false; + return held; } } @@ -127,6 +136,9 @@ 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.Information, Message = "Work item {Id} was cancelled at an admin's request.")] + public static partial void Cancelled(ILogger logger, Guid id); + [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); From 9a17b11a9fdad33edcc1dcb837020ec3671ffd2c Mon Sep 17 00:00:00 2001 From: bitbiter-dev Date: Wed, 21 Oct 2026 13:26:22 +1300 Subject: [PATCH 2/4] docs: Leave the unit suite to the pre-push hook Co-Authored-By: Claude Opus 5.5 --- CLAUDE.md | 2 ++ 1 file changed, 2 insertions(+) diff --git a/CLAUDE.md b/CLAUDE.md index bdc9325..5b43724 100644 --- a/CLAUDE.md +++ b/CLAUDE.md @@ -47,6 +47,8 @@ dotnet test --filter "FullyQualifiedName=Namespace.ClassName.MethodName" An integration test project, or a filter matching integration tests, runs nothing without `-p:TestSuite=Integration` (or `All`). +**The git hooks are the pre-commit and pre-push gates; don't run them by hand as well.** `pre-commit` runs `dotnet format --verify-no-changes` and a full build, and `pre-push` runs the whole unit suite (`.husky/task-runner.json`). While working, run the test projects or tests you touched. Don't run the full unit suite before committing or pushing; `git push` runs it, and a failure stops the push. No hook runs the integration tests or mutation testing, so run those yourself when your change needs them. + Test projects are recognised by name: `*.Tests.Unit` for unit tests and `*.Tests.Integration` for integration tests. `Directory.Build.props` applies the test SDK to both: xUnit v3, **NSubstitute** (not Moq), FluentAssertions, `System.IO.Abstractions.TestingHelpers`, and coverlet. `Directory.Build.targets` makes only the projects in the selected `TestSuite` (`Unit` by default, `Integration` or `All`) test projects, so `dotnet test` and Stryker never run the integration suite. Internal members are exposed to test projects via `InternalsVisibleTo`. **Unit or integration?** A test belongs in `Anichron.Tests.Integration` only when its subject is behaviour that only real PostgreSQL has: `ExecuteUpdate`/`ExecuteDelete`, transactions, advisory locks, row locking, constraint and partial-index enforcement. Everything else is a unit test, including a query's SQL shape (assert on `ToQueryString()` with `AnichronDbContexts.NpgsqlModel()`). The integration project starts one PostgreSQL container per run (`PostgresServer`, the same image as `src/docker-compose.yml`). Each test gets its own database, from `CreateMigratedDatabaseAsync` (real migrations) or `CreateDatabaseAsync` (empty). Its CI job is `Integration tests`, and it runs outside mutation testing, so the members it covers keep their `// Stryker disable all` and ADR 0005 rows. From 846e2e5f50844a466f38d59f290a2eae3f0a8580 Mon Sep 17 00:00:00 2001 From: bitbiter-dev Date: Wed, 21 Oct 2026 15:15:02 +1300 Subject: [PATCH 3/4] refactor(worker): Read the cancel request in the heartbeat's own UPDATE (#280) Co-Authored-By: Claude Opus 5.5 --- CLAUDE.md | 4 +-- .../Endpoints/WorkItemEndpointsHttpTests.cs | 2 +- .../Endpoints/WorkItemEndpointsTests.cs | 2 +- .../Endpoints/WorkItemEventsTests.cs | 2 +- src/Anichron.Core/Work/WorkItemReader.cs | 2 +- src/Anichron.Worker/Work/EfWorkItemStore.cs | 34 +++++++++++-------- src/Anichron.Worker/Work/WorkExecutor.cs | 6 ++-- 7 files changed, 28 insertions(+), 24 deletions(-) diff --git a/CLAUDE.md b/CLAUDE.md index 5b43724..236055b 100644 --- a/CLAUDE.md +++ b/CLAUDE.md @@ -201,8 +201,8 @@ Durable work for the Worker lives in `WorkItems` (#276, ADR 0008). Its kinds are - **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). - **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. `UpdatedAt` is the change cursor, so a heartbeat moves it only when the snapshot changed; `LeaseUntil` is what says the Worker is alive. 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. -- **Cancelling:** `POST /api/v1/admin/work-items/{id}/cancel` (#280) goes through `EfWorkItemCanceller`, one status-guarded `UPDATE`: a queued item ends `Cancelled` at once (`200`), a running one gets `CancelRequestedAt` (`202`), a finished one is `409` and an unknown id `404`. The heartbeat reads the request and cancels the handler but keeps beating until it returns; the item then ends `Cancelled` with its last progress snapshot (and a summary, if the handler returned one), shown as partial (`ResultIsPartial`). A handler that fails after the request still fails. A requested item is never queued again: release and lease recovery end it `Cancelled`, even at the stall limit. 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. +- **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. `UpdatedAt` is the change cursor, so a heartbeat moves it only when the snapshot changed; `LeaseUntil` is what says the Worker is alive. 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. +- **Cancelling:** `POST /api/v1/admin/work-items/{id}/cancel` (#280) goes through `EfWorkItemCanceller`, one status-guarded `UPDATE`: a queued item ends `Cancelled` at once (`200`), a running one gets `CancelRequestedAt` (`202`), a finished one is `409` and an unknown id `404`. The heartbeat reads the request and cancels the handler but keeps beating until it returns; the item then ends `Cancelled` with its last progress snapshot (and a summary, if the handler returned one), shown as partial (`ResultIsPartial`). A handler that fails after the request still fails. A requested item is never queued again: release and lease recovery end it `Cancelled`, even at the stall limit. - **Admin view (#279):** `GET /api/v1/admin/work-items` (filtered by status and kind, newest request first, paged), `.../{id}`, and `.../events`, a Server-Sent Events stream that re-queries `UpdatedAt` every `WorkItemEvents:IntervalSeconds` (2 s). An event's id is the cursor, and the list carries the cursor read before its page, so a client that lists and then subscribes with it as `Last-Event-ID` misses nothing; an item may arrive twice and is merged by id. The API reads items only through `IWorkItemReader` (Core), as `WorkItemView`. A result is shown once an item succeeded or was cancelled (then marked partial), an error once it failed or was cancelled. - **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`. diff --git a/src/Anichron.API.Tests.Unit/Endpoints/WorkItemEndpointsHttpTests.cs b/src/Anichron.API.Tests.Unit/Endpoints/WorkItemEndpointsHttpTests.cs index 677ff7e..e3aeddd 100644 --- a/src/Anichron.API.Tests.Unit/Endpoints/WorkItemEndpointsHttpTests.cs +++ b/src/Anichron.API.Tests.Unit/Endpoints/WorkItemEndpointsHttpTests.cs @@ -26,7 +26,7 @@ public sealed class WorkItemEndpointsHttpTests private static WorkItemView Succeeded => new(ItemId, WorkKind.Crawl, null, null, Start, WorkItemStatus.Succeeded, 1, 0, "all@host", Start, Start + Duration.FromSeconds(9), Start + Duration.FromSeconds(9), new WorkProgress(8, 8, "8 found"), - /*lang=json,strict*/ """{"found": 8}""", null); + /*lang=json,strict*/ """{"found": 8}""", null, null); private Task StartAsync() => AuthenticatedTestApp.StartAsync(api => api.MapWorkItemEndpoints(), services => diff --git a/src/Anichron.API.Tests.Unit/Endpoints/WorkItemEndpointsTests.cs b/src/Anichron.API.Tests.Unit/Endpoints/WorkItemEndpointsTests.cs index d4d04fb..ad269ce 100644 --- a/src/Anichron.API.Tests.Unit/Endpoints/WorkItemEndpointsTests.cs +++ b/src/Anichron.API.Tests.Unit/Endpoints/WorkItemEndpointsTests.cs @@ -26,7 +26,7 @@ private static WorkItemView View( string? result = /*lang=json,strict*/ """{"found": 8}""", string? error = "IOException: the root is gone") => new(ItemId, WorkKind.Crawl, ConfigId, AdminId, At(1), status, 2, 1, "all@host", - At(2), At(9), At(9), new WorkProgress(7, 8, "7 found"), result, error); + At(2), At(9), At(9), new WorkProgress(7, 8, "7 found"), result, error, null); private void Pages(int total = 1, Instant? latest = null) { diff --git a/src/Anichron.API.Tests.Unit/Endpoints/WorkItemEventsTests.cs b/src/Anichron.API.Tests.Unit/Endpoints/WorkItemEventsTests.cs index ec3a252..a843844 100644 --- a/src/Anichron.API.Tests.Unit/Endpoints/WorkItemEventsTests.cs +++ b/src/Anichron.API.Tests.Unit/Endpoints/WorkItemEventsTests.cs @@ -41,7 +41,7 @@ public WorkItemEventsTests() private static WorkItemView Changed(Guid id, int updatedAt) => new(id, WorkKind.Crawl, null, null, Start, WorkItemStatus.Running, 1, 0, "all@host", - Start, null, At(updatedAt), new WorkProgress(3, null, "3 found"), null, null); + Start, null, At(updatedAt), new WorkProgress(3, null, "3 found"), null, null, null); private void Stream(string? lastEventId = null, double intervalSeconds = 2) { diff --git a/src/Anichron.Core/Work/WorkItemReader.cs b/src/Anichron.Core/Work/WorkItemReader.cs index 803bc36..9984e81 100644 --- a/src/Anichron.Core/Work/WorkItemReader.cs +++ b/src/Anichron.Core/Work/WorkItemReader.cs @@ -21,7 +21,7 @@ public sealed record WorkItemView( WorkProgress Progress, string? Result, string? Error, - Instant? CancelRequestedAt = null); + Instant? CancelRequestedAt); public sealed record WorkItemQuery(WorkItemStatus? Status, WorkKind? Kind, int Skip, int Take); diff --git a/src/Anichron.Worker/Work/EfWorkItemStore.cs b/src/Anichron.Worker/Work/EfWorkItemStore.cs index 0b88aeb..d437a1d 100644 --- a/src/Anichron.Worker/Work/EfWorkItemStore.cs +++ b/src/Anichron.Worker/Work/EfWorkItemStore.cs @@ -48,21 +48,25 @@ public async Task HeartbeatAsync( WorkAttempt attempt, WorkProgress progress, Instant now, Duration lease, CancellationToken ct) { await using var db = await contexts.CreateDbContextAsync(ct); - var rows = await Running(db, attempt).ExecuteUpdateAsync(s => s - .SetProperty(w => w.LeaseUntil, now + lease) - .SetProperty(w => w.ProgressDone, progress.Done) - .SetProperty(w => w.ProgressTotal, progress.Total) - .SetProperty(w => w.ProgressNote, progress.Note) - .SetProperty( - w => w.UpdatedAt, - w => w.ProgressDone == progress.Done && w.ProgressTotal == progress.Total && w.ProgressNote == progress.Note - ? w.UpdatedAt - : now), ct); - if (rows == 0) - return HeartbeatOutcome.Lost; - return await Running(db, attempt).AnyAsync(w => w.CancelRequestedAt != null, ct) - ? HeartbeatOutcome.CancelRequested - : HeartbeatOutcome.Held; + var leaseUntil = now + lease; + var cancelRequested = await db.Database.SqlQuery($""" + UPDATE "WorkItems" + SET "LeaseUntil" = {leaseUntil}, "ProgressDone" = {progress.Done}, + "ProgressTotal" = CAST({progress.Total} AS bigint), "ProgressNote" = CAST({progress.Note} AS text), + "UpdatedAt" = CASE + WHEN "ProgressDone" = {progress.Done} + AND "ProgressTotal" IS NOT DISTINCT FROM CAST({progress.Total} AS bigint) + AND "ProgressNote" IS NOT DISTINCT FROM CAST({progress.Note} AS text) + THEN "UpdatedAt" ELSE {now} END + WHERE "Id" = {attempt.Id} AND "Attempt" = {attempt.Attempt} AND "Status" = 'Running' + RETURNING "CancelRequestedAt" IS NOT NULL AS "Value" + """).ToListAsync(ct); + return cancelRequested switch + { + [] => HeartbeatOutcome.Lost, + [true] => HeartbeatOutcome.CancelRequested, + _ => HeartbeatOutcome.Held, + }; } // Stryker disable all diff --git a/src/Anichron.Worker/Work/WorkExecutor.cs b/src/Anichron.Worker/Work/WorkExecutor.cs index 14845fc..33b8fb5 100644 --- a/src/Anichron.Worker/Work/WorkExecutor.cs +++ b/src/Anichron.Worker/Work/WorkExecutor.cs @@ -98,7 +98,7 @@ private async Task HeartbeatAsync( WorkAttempt attempt, WorkProgressSink progress, CancellationTokenSource running, CancellationToken done) { var interval = TimeSpan.FromSeconds(settings.WorkHeartbeatSeconds); - var held = HeartbeatOutcome.Held; + var outcome = HeartbeatOutcome.Held; try { while (true) @@ -111,7 +111,7 @@ private async Task HeartbeatAsync( await running.CancelAsync(); if (beat == HeartbeatOutcome.Lost) return beat; - held = beat; + outcome = beat; } catch (Exception ex) { @@ -121,7 +121,7 @@ private async Task HeartbeatAsync( } catch (OperationCanceledException) when (done.IsCancellationRequested) { - return held; + return outcome; } } From 570339eb385fc0df9c206b25ccb301156f58bae5 Mon Sep 17 00:00:00 2001 From: bitbiter-dev Date: Wed, 21 Oct 2026 18:12:46 +1300 Subject: [PATCH 4/4] refactor: Guard work-item writes by entity dependency; document cancel due-ness (#280) Co-Authored-By: Claude Opus 5.5 --- CLAUDE.md | 4 ++-- docs/schema.sql | 2 +- .../ArchitectureRules.cs | 9 +++++---- src/Anichron.Core/Data/AnichronDbContext.cs | 3 ++- ...ner.cs => 20261021051155_CancelWorkItems.Designer.cs} | 4 ++-- ...celWorkItems.cs => 20261021051155_CancelWorkItems.cs} | 2 +- .../Migrations/AnichronDbContextModelSnapshot.cs | 2 +- 7 files changed, 14 insertions(+), 12 deletions(-) rename src/Anichron.Core/Migrations/{20261020234633_CancelWorkItems.Designer.cs => 20261021051155_CancelWorkItems.Designer.cs} (99%) rename src/Anichron.Core/Migrations/{20261020234633_CancelWorkItems.cs => 20261021051155_CancelWorkItems.cs} (96%) diff --git a/CLAUDE.md b/CLAUDE.md index 236055b..cac8811 100644 --- a/CLAUDE.md +++ b/CLAUDE.md @@ -102,7 +102,7 @@ Four projects with strict separation of concerns: **Write ownership is split by domain, not centralized.** - **Worker is the only writer of *media* data** — `MediaAsset`, `Metadata`, `ProxyFile`, `Burst`, plus all reconciliation and soft-deletes happen here. The API never writes media: outside its DI wiring it never touches `AnichronDbContext` (architecture rule `api-reaches-data-through-repositories`), and the media write repository stays out of it (`only-worker-writes-media`). Its one identity-side write is the `TokenCleanup` work item's handler deleting expired refresh tokens. - **API owns all *identity/auth/admin* writes** — registration, login/refresh-token issuance and revocation, lockout state, invites, and admin user/storage-config CRUD all persist through API services (`AuthService`, `AdminUserService`, `AdminStorageConfigService`, `TokenService`, etc.) via `IUnitOfWork.SaveChangesAsync`. The API is read-only *only with respect to media*. The one exception is `AssetInteraction` (star/like/hide, #46): per-user state, written by the API through `IAssetInteractionWriter` (Core), which takes and returns plain records so no API type touches the entity. It checks only `VisibleTo` (never `NotHiddenFor`, so a hidden asset can be unhidden), and a first write that loses the `(UserId, AssetId)` unique race reloads and reapplies once. -- **Work items** are inserted only through `IWorkQueue` (architecture rule `only-work-queue-inserts-work-items`); the API's only other write is an admin's cancel through `IWorkItemCanceller` (`api-writes-work-items-only-by-cancel`: outside the Worker, only `EfWorkQueue`, `EfWorkItemReader` and `EfWorkItemCanceller` touch `WorkItems`); every other state change belongs to the Worker's executor, and handlers live only in the Worker (`work-handlers-live-in-worker`). See "Work queue" below. +- **Work items** are inserted only through `IWorkQueue` (architecture rule `only-work-queue-inserts-work-items`); the API's only other write is an admin's cancel through `IWorkItemCanceller` (`work-items-change-only-through-queue-or-cancel`: outside the Worker, only `EfWorkQueue`, `EfWorkItemReader` and `EfWorkItemCanceller` depend on the `WorkItem` entity; raw SQL naming the table is beyond the rule and is for review); every other state change belongs to the Worker's executor, and handlers live only in the Worker (`work-handlers-live-in-worker`). See "Work queue" below. - Both hosts apply EF migrations at startup, serialised by a PostgreSQL advisory lock (`MigrateWithAdvisoryLockAsync`). A planned `Anichron.UI` (NextJS/React) for the Instagram Story-style flashback experience is documented but not yet in the solution. @@ -202,7 +202,7 @@ Durable work for the Worker lives in `WorkItems` (#276, ADR 0008). Its kinds are - **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. `UpdatedAt` is the change cursor, so a heartbeat moves it only when the snapshot changed; `LeaseUntil` is what says the Worker is alive. 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. -- **Cancelling:** `POST /api/v1/admin/work-items/{id}/cancel` (#280) goes through `EfWorkItemCanceller`, one status-guarded `UPDATE`: a queued item ends `Cancelled` at once (`200`), a running one gets `CancelRequestedAt` (`202`), a finished one is `409` and an unknown id `404`. The heartbeat reads the request and cancels the handler but keeps beating until it returns; the item then ends `Cancelled` with its last progress snapshot (and a summary, if the handler returned one), shown as partial (`ResultIsPartial`). A handler that fails after the request still fails. A requested item is never queued again: release and lease recovery end it `Cancelled`, even at the stall limit. +- **Cancelling:** `POST /api/v1/admin/work-items/{id}/cancel` (#280) goes through `EfWorkItemCanceller`, one status-guarded `UPDATE`: a queued item ends `Cancelled` at once (`200`), a running one gets `CancelRequestedAt` (`202`), a finished one is `409` and an unknown id `404`. The heartbeat reads the request and cancels the handler but keeps beating until it returns; the item then ends `Cancelled` with its last progress snapshot (and a summary, if the handler returned one), shown as partial (`ResultIsPartial`). A handler that fails after the request still fails. A requested item is never queued again: release and lease recovery end it `Cancelled`, even at the stall limit. A cancelled periodic item counts as that run for due-ness, so the next one follows a full interval later; were it not counted, the scheduler would enqueue it again at the next poll and undo the cancel. - **Admin view (#279):** `GET /api/v1/admin/work-items` (filtered by status and kind, newest request first, paged), `.../{id}`, and `.../events`, a Server-Sent Events stream that re-queries `UpdatedAt` every `WorkItemEvents:IntervalSeconds` (2 s). An event's id is the cursor, and the list carries the cursor read before its page, so a client that lists and then subscribes with it as `Last-Event-ID` misses nothing; an item may arrive twice and is merged by id. The API reads items only through `IWorkItemReader` (Core), as `WorkItemView`. A result is shown once an item succeeded or was cancelled (then marked partial), an error once it failed or was cancelled. - **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`. diff --git a/docs/schema.sql b/docs/schema.sql index 1365bd9..1171623 100644 --- a/docs/schema.sql +++ b/docs/schema.sql @@ -97,7 +97,7 @@ COMMENT ON COLUMN "WorkItems"."Stalls" IS 'Lease expiries, counted by lease reco COMMENT ON COLUMN "WorkItems"."ClaimedBy" IS 'The claiming Worker for display: its WORKER__USER or ''all'', then its host name.'; COMMENT ON COLUMN "WorkItems"."LeaseUntil" IS 'Extended by every heartbeat. A Running item past it belongs to a dead or paused Worker and goes back to the queue.'; COMMENT ON COLUMN "WorkItems"."UpdatedAt" IS 'Moves on every insert, claim and transition, and on a heartbeat only when the progress snapshot changed, so a reader can ask what changed since a point in time.'; -COMMENT ON COLUMN "WorkItems"."CancelRequestedAt" IS 'When an admin asked to cancel the item. A Queued item is cancelled at once; a Running one keeps running until the Worker''s next heartbeat reads this and cancels its handler. A requested item is never queued again: release and lease recovery end it Cancelled.'; +COMMENT ON COLUMN "WorkItems"."CancelRequestedAt" IS 'When an admin asked to cancel the item. A Queued item is cancelled at once; a Running one keeps running until the Worker''s next heartbeat reads this and cancels its handler. A requested item is never queued again: release and lease recovery end it Cancelled. A cancelled periodic item counts as that run, so the next one follows a full interval later.'; COMMENT ON COLUMN "WorkItems"."ProgressNote" IS 'The latest progress snapshot (done, optional total, note), overwritten by each heartbeat; no history is kept.'; COMMENT ON COLUMN "WorkItems"."Result" IS 'The handler''s results summary, written when the item succeeds, or when it is cancelled after its handler returned one; a Cancelled item''s results are partial.'; diff --git a/src/Anichron.Architecture.Tests.Unit/ArchitectureRules.cs b/src/Anichron.Architecture.Tests.Unit/ArchitectureRules.cs index a4ed0ea..198faac 100644 --- a/src/Anichron.Architecture.Tests.Unit/ArchitectureRules.cs +++ b/src/Anichron.Architecture.Tests.Unit/ArchitectureRules.cs @@ -28,7 +28,7 @@ internal static class ArchitectureRules public const string DomainUsesNodaTime = "domain-uses-nodatime"; public const string OnlyWorkQueueInsertsWorkItems = "only-work-queue-inserts-work-items"; public const string WorkHandlersLiveInWorker = "work-handlers-live-in-worker"; - public const string ApiWritesWorkItemsOnlyByCancel = "api-writes-work-items-only-by-cancel"; + public const string WorkItemsChangeOnlyThroughQueueOrCancel = "work-items-change-only-through-queue-or-cancel"; private const string Domain = "Anichron.Core.Domain"; private const string ApiWiring = "Anichron.API.Infrastructure"; @@ -84,13 +84,14 @@ internal static class ArchitectureRules new(WorkHandlersLiveInWorker, Types().That().ImplementInterface(typeof(IWorkHandler<>)) .Should().ResideInAssembly(WorkerAssembly)), - new(ApiWritesWorkItemsOnlyByCancel, + new(WorkItemsChangeOnlyThroughQueueOrCancel, Types().That().ResideInAssembly(ApiAssembly, CoreAssembly, InfrastructureAssembly) + .And().DoNotResideInNamespace(Domain) + .And().DoNotHaveFullName(typeof(AnichronDbContext).FullName) .And().DoNotHaveFullName(WorkQueue) .And().DoNotHaveFullName(WorkItemReader) .And().DoNotHaveFullName(WorkItemCanceller) - .Should().NotCallAny(MethodMembers().That().AreDeclaredIn(typeof(AnichronDbContext)) - .And().HaveNameMatching(@"^get_WorkItems\("))), + .Should().NotDependOnAny(typeof(WorkItem))), ]; public static ArchitectureRule Named(string name) => All.Single(rule => rule.Name == name); diff --git a/src/Anichron.Core/Data/AnichronDbContext.cs b/src/Anichron.Core/Data/AnichronDbContext.cs index 49de1f3..5d5a7f2 100644 --- a/src/Anichron.Core/Data/AnichronDbContext.cs +++ b/src/Anichron.Core/Data/AnichronDbContext.cs @@ -352,7 +352,8 @@ protected override void OnModelCreating(ModelBuilder modelBuilder) entity.Property(w => w.CancelRequestedAt).HasComment( "When an admin asked to cancel the item. A Queued item is cancelled at once; a Running " + "one keeps running until the Worker's next heartbeat reads this and cancels its handler. " - + "A requested item is never queued again: release and lease recovery end it Cancelled."); + + "A requested item is never queued again: release and lease recovery end it Cancelled. " + + "A cancelled periodic item counts as that run, so the next one follows a full interval later."); entity.Property(w => w.ProgressNote).HasComment( "The latest progress snapshot (done, optional total, note), overwritten by each " + "heartbeat; no history is kept."); diff --git a/src/Anichron.Core/Migrations/20261020234633_CancelWorkItems.Designer.cs b/src/Anichron.Core/Migrations/20261021051155_CancelWorkItems.Designer.cs similarity index 99% rename from src/Anichron.Core/Migrations/20261020234633_CancelWorkItems.Designer.cs rename to src/Anichron.Core/Migrations/20261021051155_CancelWorkItems.Designer.cs index a9df0b1..4d25cb9 100644 --- a/src/Anichron.Core/Migrations/20261020234633_CancelWorkItems.Designer.cs +++ b/src/Anichron.Core/Migrations/20261021051155_CancelWorkItems.Designer.cs @@ -13,7 +13,7 @@ namespace Anichron.Core.Migrations { [DbContext(typeof(AnichronDbContext))] - [Migration("20261020234633_CancelWorkItems")] + [Migration("20261021051155_CancelWorkItems")] partial class CancelWorkItems { /// @@ -437,7 +437,7 @@ protected override void BuildTargetModel(ModelBuilder modelBuilder) b.Property("CancelRequestedAt") .HasColumnType("timestamp with time zone") - .HasComment("When an admin asked to cancel the item. A Queued item is cancelled at once; a Running one keeps running until the Worker's next heartbeat reads this and cancels its handler. A requested item is never queued again: release and lease recovery end it Cancelled."); + .HasComment("When an admin asked to cancel the item. A Queued item is cancelled at once; a Running one keeps running until the Worker's next heartbeat reads this and cancels its handler. A requested item is never queued again: release and lease recovery end it Cancelled. A cancelled periodic item counts as that run, so the next one follows a full interval later."); b.Property("ClaimedBy") .HasColumnType("text") diff --git a/src/Anichron.Core/Migrations/20261020234633_CancelWorkItems.cs b/src/Anichron.Core/Migrations/20261021051155_CancelWorkItems.cs similarity index 96% rename from src/Anichron.Core/Migrations/20261020234633_CancelWorkItems.cs rename to src/Anichron.Core/Migrations/20261021051155_CancelWorkItems.cs index 8195b65..c419c29 100644 --- a/src/Anichron.Core/Migrations/20261020234633_CancelWorkItems.cs +++ b/src/Anichron.Core/Migrations/20261021051155_CancelWorkItems.cs @@ -32,7 +32,7 @@ protected override void Up(MigrationBuilder migrationBuilder) table: "WorkItems", type: "timestamp with time zone", nullable: true, - comment: "When an admin asked to cancel the item. A Queued item is cancelled at once; a Running one keeps running until the Worker's next heartbeat reads this and cancels its handler. A requested item is never queued again: release and lease recovery end it Cancelled."); + comment: "When an admin asked to cancel the item. A Queued item is cancelled at once; a Running one keeps running until the Worker's next heartbeat reads this and cancels its handler. A requested item is never queued again: release and lease recovery end it Cancelled. A cancelled periodic item counts as that run, so the next one follows a full interval later."); } /// diff --git a/src/Anichron.Core/Migrations/AnichronDbContextModelSnapshot.cs b/src/Anichron.Core/Migrations/AnichronDbContextModelSnapshot.cs index 5c6a02e..37a8492 100644 --- a/src/Anichron.Core/Migrations/AnichronDbContextModelSnapshot.cs +++ b/src/Anichron.Core/Migrations/AnichronDbContextModelSnapshot.cs @@ -434,7 +434,7 @@ protected override void BuildModel(ModelBuilder modelBuilder) b.Property("CancelRequestedAt") .HasColumnType("timestamp with time zone") - .HasComment("When an admin asked to cancel the item. A Queued item is cancelled at once; a Running one keeps running until the Worker's next heartbeat reads this and cancels its handler. A requested item is never queued again: release and lease recovery end it Cancelled."); + .HasComment("When an admin asked to cancel the item. A Queued item is cancelled at once; a Running one keeps running until the Worker's next heartbeat reads this and cancels its handler. A requested item is never queued again: release and lease recovery end it Cancelled. A cancelled periodic item counts as that run, so the next one follows a full interval later."); b.Property("ClaimedBy") .HasColumnType("text")