feat: Cancel queued and running work items (#280)

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.
@@ -100,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`); 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` (`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.
@@ -200,7 +202,8 @@ 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<TPayload>` 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.
+- **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<TPayload>` in the Worker registered with `AddWorkHandler`, and for a periodic kind a `WorkSchedule<TPayload>` 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`.
 

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

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. 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.';
+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" (

src/Anichron.API.Tests.Unit/Endpoints/WorkItemEndpointsHttpTests.cs

@@ -19,18 +19,20 @@ public sealed class WorkItemEndpointsHttpTests
     private static readonly Guid ItemId = Guid.Parse("aaaaaaaa-aaaa-aaaa-aaaa-aaaaaaaaaaaa");
 
     private readonly IWorkItemReader reader = Substitute.For<IWorkItemReader>();
+    private readonly IWorkItemCanceller canceller = Substitute.For<IWorkItemCanceller>();
 
     private static CancellationToken Ct => TestContext.Current.CancellationToken;
 
     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<AuthenticatedTestApp> StartAsync()
         => AuthenticatedTestApp.StartAsync(api => api.MapWorkItemEndpoints(), services =>
         {
             services.AddSingleton(reader);
+            services.AddSingleton(canceller);
             services.AddSingleton<TimeProvider>(new FakeTimeProvider());
             services.AddSingleton(Options.Create(new WorkItemEventsSettings()));
             services.AddScoped<WorkItemEvents>();
@@ -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<Guid>(), Arg.Any<CancellationToken>());
+    }
+
+    [Fact]
+    public async Task Cancel_PostedByAnAdmin_CancelsTheItemInThePathAsync()
+    {
+        canceller.CancelAsync(ItemId, Arg.Any<CancellationToken>()).Returns(CancelOutcome.Finished);
+        reader.FindAsync(ItemId, Arg.Any<CancellationToken>()).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]