diff --git a/autodoc/interfaces/proposal-operation-vocabulary.md b/autodoc/interfaces/proposal-operation-vocabulary.md index f5caf7b4c7..7aa7b91714 100644 --- a/autodoc/interfaces/proposal-operation-vocabulary.md +++ b/autodoc/interfaces/proposal-operation-vocabulary.md @@ -15,7 +15,7 @@ Card operations: - `update`: `cardId`; at least one of `title`, `description`, `dueDate`, `clearDueDate`, `labels`, or `labelIds`. At the operation-payload layer, an explicit null `dueDate` clears it. The chat tool treats `due_date: null` as omitted and emits clearing only for `clear_due_date: true`. `dueDate` and a true `clearDueDate` are mutually exclusive. - `move`: `cardId`, `columnId`. - `archive`: `cardId`. This legacy identity permanently retains Block semantics: applying it marks the card blocked with the generated reason `Archived by an approved proposal.` The preview shows that exact blocked-state transition before approval. Existing approvals and the legacy MCP `archive_card` and chat `propose_archive_card` tools keep this contract. -- `archive-lifecycle` / `restore-lifecycle`: `cardId`, `expectedUpdatedAt` (the displayed card timestamp). Archive hides the retained card from active work; restore returns it to its original column/position, subject to that column's WIP limit measured against the occupancy the same proposal's earlier operations produce, so a restore behind a card create or move that takes the last slot is refused at preview instead of failing mid-apply (`#2926`). A restore cannot sit behind another lifecycle operation: `ProposalHierarchyValidator` admits at most one hierarchy-affecting operation per proposal, so batch archive/restore in a single proposal is refused outright and the projection's lifecycle deltas are carried only against that gate being relaxed later. Preview and Apply reject stale timestamps, wrong archive state, unavailable columns, and another operation on the same card in the same proposal. The separate MCP `archive_card_lifecycle` / `restore_archived_card` tools only create proposals; explicit approval and Apply remain required. Both names are in `SideEffectAnalyzer.CardMutatingActions`, so the review Cards row discloses the board mutation, and Apply persists exactly one `Archived`/`Unarchived` audit row written by `ExecutionAuditRecorder` with the proposal provenance (the handler passes `recordLifecycleAudit: false` so `CardService.SetArchivedAsync` does not stage a second row); direct API archive/restore keeps its own single actor-stamped row (`#2939`). The realtime `card.archived`/`card.restored` event (and the `card.updated` events for children detached by an archive) is likewise staged on the apply lane: the handler passes the executor's `DeferredBoardRealtimeNotifier` as `notificationSink`, and `AutomationExecutorService` publishes the buffer only after `CommitTransactionAsync` and discards it on every non-commit exit, so an operation that fails later emits nothing; direct API archive/restore passes no sink and still notifies immediately (`#2934`); the trade is that a lifecycle event's outbound webhook delivery rows are now written by that post-commit flush, outside any transaction, so a host crash between the commit and the flush loses them with no retry, where before they rolled back with the mutation - tracked in `#3024`. +- `archive-lifecycle` / `restore-lifecycle`: `cardId`, `expectedUpdatedAt` (the displayed card timestamp). Archive hides the retained card from active work; restore returns it to its original column/position, subject to that column's WIP limit measured against the occupancy the same proposal's earlier operations produce, so a restore behind a card create or move that takes the last slot is refused at preview instead of failing mid-apply (`#2926`). A restore cannot sit behind another lifecycle operation: `ProposalHierarchyValidator` admits at most one hierarchy-affecting operation per proposal, so batch archive/restore in a single proposal is refused outright and the projection's lifecycle deltas are carried only against that gate being relaxed later. Preview and Apply reject stale timestamps, wrong archive state, unavailable columns, and another operation on the same card in the same proposal. The separate MCP `archive_card_lifecycle` / `restore_archived_card` tools only create proposals; explicit approval and Apply remain required. Both names are in `SideEffectAnalyzer.CardMutatingActions`, so the review Cards row discloses the board mutation, and Apply persists exactly one `Archived`/`Unarchived` audit row written by `ExecutionAuditRecorder` with the proposal provenance (the handler passes `recordLifecycleAudit: false` so `CardService.SetArchivedAsync` does not stage a second row); direct API archive/restore keeps its own single actor-stamped row (`#2939`). The realtime `card.archived`/`card.restored` event (and the `card.updated` events for children detached by an archive) is likewise staged on the apply lane: the handler passes the executor's `DeferredBoardRealtimeNotifier` as `notificationSink`, and `AutomationExecutorService` publishes the buffer only after `CommitTransactionAsync` and discards it on every non-commit exit, so an operation that fails later emits nothing; direct API archive/restore passes no sink and still notifies immediately (`#2934`). The executor prepares filtered outbound webhook delivery rows through `ITransactionalBoardMutationNotifier` before its final Applied-status save, so those rows and subscription trigger timestamps commit or roll back with the proposal. After commit, prepared events publish realtime only; the existing delivery worker can claim the durable rows even if the realtime flush is lost (`#3024`). This covers already-buffered proposal events and does not change immediate notification producers or the separate assignment collector. - `add-label` / `remove-label`: `cardId` plus exactly one of board-scoped `labelId` or `labelName`. Separator-free and underscore aliases remain accepted at apply time for existing callers. - `add-relation` / `remove-relation`: `boardId`, source `cardId`, target `relatedCardId`, `relationType`, and mandatory non-negative `expectedRevision`; `targetType` is `card`. The admitted input kinds are `relates-to`, `blocks`, `depends-on`, `duplicates`, and `spawned-from`. `depends-on(A, B)` is stored and previewed as `blocks(B, A)`; `relates-to` is deterministically ordered, while the other kinds retain direction. Both endpoints must be active cards on the proposal board, except for a preceding card create with a preallocated operation target ID, which is valid within the ordered proposal. A proposal contains at most one relation operation and cannot combine it with card archive-lifecycle, restore-lifecycle, or delete. Apply preserves the caller's original relation revision pin; it never restamps it after an earlier operation. - `workItemType` (`Task` / `Epic` / `Spike`) is read only by `create` and `update`; every other card action ignores it at apply, so `ProposalOperationContractValidator` rejects it there with `Parameter 'workItemType' is not supported by card action ''` rather than letting the preview announce a type transition Apply would not perform (`#2950`). diff --git a/backend/src/Taskdeck.Api/Realtime/CompositeBoardRealtimeNotifier.cs b/backend/src/Taskdeck.Api/Realtime/CompositeBoardRealtimeNotifier.cs index 220450b056..38d3ff3e01 100644 --- a/backend/src/Taskdeck.Api/Realtime/CompositeBoardRealtimeNotifier.cs +++ b/backend/src/Taskdeck.Api/Realtime/CompositeBoardRealtimeNotifier.cs @@ -2,7 +2,7 @@ namespace Taskdeck.Api.Realtime; -public sealed class CompositeBoardRealtimeNotifier : IBoardRealtimeNotifier +public sealed class CompositeBoardRealtimeNotifier : IBoardRealtimeNotifier, ITransactionalBoardMutationNotifier { private readonly SignalRBoardRealtimeNotifier _signalRNotifier; private readonly WebhookBoardMutationNotifier _webhookNotifier; @@ -35,6 +35,20 @@ await NotifySafeAsync( ct => _webhookNotifier.NotifyBoardMutationAsync(mutation, ct)); } + public Task StageBoardMutationAsync( + BoardRealtimeEvent mutation, + CancellationToken cancellationToken = default) => + _webhookNotifier.StageBoardMutationAsync(mutation, cancellationToken); + + public Task NotifyCommittedBoardMutationAsync( + BoardRealtimeEvent mutation, + CancellationToken cancellationToken = default) => + NotifySafeAsync( + "signalr", + mutation, + cancellationToken, + ct => _signalRNotifier.NotifyBoardMutationAsync(mutation, ct)); + private async Task NotifySafeAsync( string channel, BoardRealtimeEvent mutation, diff --git a/backend/src/Taskdeck.Api/Realtime/WebhookBoardMutationNotifier.cs b/backend/src/Taskdeck.Api/Realtime/WebhookBoardMutationNotifier.cs index 3e4e74b4f1..f545386a2b 100644 --- a/backend/src/Taskdeck.Api/Realtime/WebhookBoardMutationNotifier.cs +++ b/backend/src/Taskdeck.Api/Realtime/WebhookBoardMutationNotifier.cs @@ -42,4 +42,16 @@ public async Task NotifyBoardMutationAsync( mutation.BoardId); } } + + public async Task StageBoardMutationAsync( + BoardRealtimeEvent mutation, + CancellationToken cancellationToken = default) + { + var result = await _outboundWebhookService.StageBoardMutationAsync(mutation, cancellationToken); + if (!result.IsSuccess) + { + throw new InvalidOperationException( + $"Could not stage outbound webhook deliveries ({result.ErrorCode})."); + } + } } diff --git a/backend/src/Taskdeck.Application/Services/AutomationExecutorService.cs b/backend/src/Taskdeck.Application/Services/AutomationExecutorService.cs index 82a0bb2a60..ad03e5e313 100644 --- a/backend/src/Taskdeck.Application/Services/AutomationExecutorService.cs +++ b/backend/src/Taskdeck.Application/Services/AutomationExecutorService.cs @@ -397,6 +397,12 @@ await _auditRecorder.RecordAsync(operation, effectiveProposal, cancellationToken return Result.Failure(failedResult.ErrorCode, failureReason); } + // Prepare durable webhook deliveries while the operation transaction is still open. + // The Applied status save below persists the board writes, audit rows, status, and + // prepared Pending deliveries together. Preparation failures must abort all of them. + if (deferredNotifications is not null) + await deferredNotifications.PrepareAsync(cancellationToken); + // The board marker, operation effects, audit rows, and Applied status share this outer // transaction. Do not re-check archived state here: an approved operation may itself // archive the board, and the pre-operation guard already ordered that legitimate write. diff --git a/backend/src/Taskdeck.Application/Services/DeferredBoardRealtimeNotifier.cs b/backend/src/Taskdeck.Application/Services/DeferredBoardRealtimeNotifier.cs index 693fa2a921..49401e4c20 100644 --- a/backend/src/Taskdeck.Application/Services/DeferredBoardRealtimeNotifier.cs +++ b/backend/src/Taskdeck.Application/Services/DeferredBoardRealtimeNotifier.cs @@ -21,6 +21,7 @@ public sealed class DeferredBoardRealtimeNotifier : IBoardRealtimeNotifier { private readonly IBoardRealtimeNotifier _inner; private readonly List _pending = new(); + private int _preparedCount; public DeferredBoardRealtimeNotifier(IBoardRealtimeNotifier? inner = null) { @@ -30,6 +31,9 @@ public DeferredBoardRealtimeNotifier(IBoardRealtimeNotifier? inner = null) /// Events staged but not yet published. Diagnostics and tests only. public int PendingCount => _pending.Count; + /// Events whose durable channel has been prepared in the caller's transaction. + public int PreparedCount => _preparedCount; + public Task NotifyBoardMutationAsync( BoardRealtimeEvent mutation, CancellationToken cancellationToken = default) @@ -38,6 +42,23 @@ public Task NotifyBoardMutationAsync( return Task.CompletedTask; } + /// + /// Prepares the durable channel for every pending event without draining the buffer. Sinks + /// without the supplemental transactional capability retain the legacy flush behavior. + /// Successfully prepared events are remembered so a repeated call cannot stage duplicates. + /// + public async Task PrepareAsync(CancellationToken cancellationToken = default) + { + if (_inner is not ITransactionalBoardMutationNotifier transactional) + return; + + while (_preparedCount < _pending.Count) + { + await transactional.StageBoardMutationAsync(_pending[_preparedCount], cancellationToken); + _preparedCount++; + } + } + /// /// Publishes every staged event downstream, in staging order, and empties the buffer. /// The buffer is emptied before the first publish, so a downstream failure can never @@ -49,11 +70,22 @@ public async Task FlushAsync(CancellationToken cancellationToken = default) return; var batch = _pending.ToArray(); + var preparedCount = _preparedCount; _pending.Clear(); - foreach (var mutation in batch) - await _inner.NotifyBoardMutationAsync(mutation, cancellationToken); + _preparedCount = 0; + for (var index = 0; index < batch.Length; index++) + { + if (index < preparedCount && _inner is ITransactionalBoardMutationNotifier transactional) + await transactional.NotifyCommittedBoardMutationAsync(batch[index], cancellationToken); + else + await _inner.NotifyBoardMutationAsync(batch[index], cancellationToken); + } } /// Drops every staged event — the write they describe did not survive. - public void Discard() => _pending.Clear(); + public void Discard() + { + _pending.Clear(); + _preparedCount = 0; + } } diff --git a/backend/src/Taskdeck.Application/Services/IOutboundWebhookService.cs b/backend/src/Taskdeck.Application/Services/IOutboundWebhookService.cs index b69c92c81e..5dfeb94135 100644 --- a/backend/src/Taskdeck.Application/Services/IOutboundWebhookService.cs +++ b/backend/src/Taskdeck.Application/Services/IOutboundWebhookService.cs @@ -30,4 +30,8 @@ Task RevokeSubscriptionAsync( Task EnqueueBoardMutationAsync( BoardRealtimeEvent mutation, CancellationToken cancellationToken = default); + + Task StageBoardMutationAsync( + BoardRealtimeEvent mutation, + CancellationToken cancellationToken = default); } diff --git a/backend/src/Taskdeck.Application/Services/ITransactionalBoardMutationNotifier.cs b/backend/src/Taskdeck.Application/Services/ITransactionalBoardMutationNotifier.cs new file mode 100644 index 0000000000..734eb30f43 --- /dev/null +++ b/backend/src/Taskdeck.Application/Services/ITransactionalBoardMutationNotifier.cs @@ -0,0 +1,16 @@ +namespace Taskdeck.Application.Services; + +/// +/// Optional capability for notification sinks that can prepare durable work inside a caller-owned +/// transaction, then publish only the non-durable channel after that transaction commits. +/// +public interface ITransactionalBoardMutationNotifier +{ + Task StageBoardMutationAsync( + BoardRealtimeEvent mutation, + CancellationToken cancellationToken = default); + + Task NotifyCommittedBoardMutationAsync( + BoardRealtimeEvent mutation, + CancellationToken cancellationToken = default); +} diff --git a/backend/src/Taskdeck.Application/Services/OutboundWebhookService.cs b/backend/src/Taskdeck.Application/Services/OutboundWebhookService.cs index b08d993ff6..41013c671a 100644 --- a/backend/src/Taskdeck.Application/Services/OutboundWebhookService.cs +++ b/backend/src/Taskdeck.Application/Services/OutboundWebhookService.cs @@ -163,6 +163,20 @@ public async Task RevokeSubscriptionAsync( public async Task EnqueueBoardMutationAsync( BoardRealtimeEvent mutation, CancellationToken cancellationToken = default) + { + var stageResult = await StageBoardMutationAsync(mutation, cancellationToken); + if (!stageResult.IsSuccess) + { + return stageResult; + } + + await _unitOfWork.SaveChangesAsync(cancellationToken); + return Result.Success(); + } + + public async Task StageBoardMutationAsync( + BoardRealtimeEvent mutation, + CancellationToken cancellationToken = default) { var eventType = $"{mutation.EntityType}.{mutation.Operation}".Trim().ToLowerInvariant(); if (string.IsNullOrWhiteSpace(eventType) || eventType == ".") @@ -202,7 +216,6 @@ public async Task EnqueueBoardMutationAsync( subscription.MarkTriggered(); } - await _unitOfWork.SaveChangesAsync(cancellationToken); return Result.Success(); } diff --git a/backend/tests/Taskdeck.Api.Tests/CompositeBoardRealtimeNotifierTests.cs b/backend/tests/Taskdeck.Api.Tests/CompositeBoardRealtimeNotifierTests.cs index 38321d2118..94dc8d7707 100644 --- a/backend/tests/Taskdeck.Api.Tests/CompositeBoardRealtimeNotifierTests.cs +++ b/backend/tests/Taskdeck.Api.Tests/CompositeBoardRealtimeNotifierTests.cs @@ -175,6 +175,53 @@ public async Task NotifyBoardMutationAsync_ShouldNotThrow_WhenBothChannelsFail() webhookLogger.Entries.Should().ContainSingle(e => e.Level == LogLevel.Error); } + [Fact] + public async Task StageBoardMutationAsync_ShouldPrepareWebhookOnly_AndPropagateFailure() + { + var mutation = CreateMutation(); + var signalRClientProxy = new RecordingClientProxy(); + var outboundService = new RecordingOutboundWebhookService(); + var notifier = CreateNotifier(signalRClientProxy, outboundService); + + await notifier.StageBoardMutationAsync(mutation, CancellationToken.None); + + outboundService.StageCalls.Should().ContainSingle(); + outboundService.EnqueueCalls.Should().BeEmpty(); + signalRClientProxy.MethodName.Should().BeNull(); + + outboundService.StageResultToReturn = Result.Failure("stage_failed", "queue failed"); + var act = () => notifier.StageBoardMutationAsync(CreateMutation(), CancellationToken.None); + await act.Should().ThrowAsync(); + } + + [Fact] + public async Task NotifyCommittedBoardMutationAsync_ShouldPublishSignalROnly() + { + var mutation = CreateMutation(); + var signalRClientProxy = new RecordingClientProxy(); + var outboundService = new RecordingOutboundWebhookService(); + var notifier = CreateNotifier(signalRClientProxy, outboundService); + + await notifier.NotifyCommittedBoardMutationAsync(mutation, CancellationToken.None); + + signalRClientProxy.MethodName.Should().Be("boardMutation"); + outboundService.StageCalls.Should().BeEmpty(); + outboundService.EnqueueCalls.Should().BeEmpty(); + } + + private static CompositeBoardRealtimeNotifier CreateNotifier( + RecordingClientProxy signalRClientProxy, + RecordingOutboundWebhookService outboundService) + { + var hubContext = new FakeHubContext(signalRClientProxy); + return new CompositeBoardRealtimeNotifier( + new SignalRBoardRealtimeNotifier(hubContext), + new WebhookBoardMutationNotifier( + outboundService, + new InMemoryLogger()), + new InMemoryLogger()); + } + private static BoardRealtimeEvent CreateMutation() { return new BoardRealtimeEvent( @@ -187,8 +234,11 @@ private static BoardRealtimeEvent CreateMutation() private sealed class RecordingOutboundWebhookService : IOutboundWebhookService { - public List<(BoardRealtimeEvent Mutation, CancellationToken CancellationToken)> Calls { get; } = []; + public List<(BoardRealtimeEvent Mutation, CancellationToken CancellationToken)> EnqueueCalls { get; } = []; + public List<(BoardRealtimeEvent Mutation, CancellationToken CancellationToken)> StageCalls { get; } = []; + public List<(BoardRealtimeEvent Mutation, CancellationToken CancellationToken)> Calls => EnqueueCalls; public Result ResultToReturn { get; set; } = Result.Success(); + public Result StageResultToReturn { get; set; } = Result.Success(); public Exception? ExceptionToThrow { get; set; } public Task> CreateSubscriptionAsync( @@ -229,7 +279,7 @@ public Task EnqueueBoardMutationAsync( BoardRealtimeEvent mutation, CancellationToken cancellationToken = default) { - Calls.Add((mutation, cancellationToken)); + EnqueueCalls.Add((mutation, cancellationToken)); if (ExceptionToThrow is not null) { @@ -238,6 +288,18 @@ public Task EnqueueBoardMutationAsync( return Task.FromResult(ResultToReturn); } + + public Task StageBoardMutationAsync( + BoardRealtimeEvent mutation, + CancellationToken cancellationToken = default) + { + StageCalls.Add((mutation, cancellationToken)); + + if (ExceptionToThrow is not null) + throw ExceptionToThrow; + + return Task.FromResult(StageResultToReturn); + } } private sealed class FakeHubContext : IHubContext diff --git a/backend/tests/Taskdeck.Api.Tests/ProposalWebhookDurabilityApiTests.cs b/backend/tests/Taskdeck.Api.Tests/ProposalWebhookDurabilityApiTests.cs new file mode 100644 index 0000000000..d716bb61b9 --- /dev/null +++ b/backend/tests/Taskdeck.Api.Tests/ProposalWebhookDurabilityApiTests.cs @@ -0,0 +1,350 @@ +using System.Net; +using System.Net.Http.Headers; +using System.Net.Http.Json; +using FluentAssertions; +using Microsoft.AspNetCore.Mvc.Testing; +using Microsoft.AspNetCore.TestHost; +using Microsoft.EntityFrameworkCore; +using Microsoft.EntityFrameworkCore.Diagnostics; +using Microsoft.Extensions.Configuration; +using Microsoft.Extensions.DependencyInjection; +using Microsoft.Extensions.DependencyInjection.Extensions; +using Taskdeck.Api.Tests.Support; +using Taskdeck.Application.Services; +using Taskdeck.Application.DTOs; +using Taskdeck.Domain.Entities; +using Taskdeck.Domain.Enums; +using Taskdeck.Infrastructure; +using Taskdeck.Infrastructure.Persistence; +using Xunit; + +namespace Taskdeck.Api.Tests; + +/// +/// Exercises the proposal executor against the real SQLite queue. The workerless factory is +/// deliberate: these tests inspect pending deliveries at the commit-to-notification hand-off and +/// must never race a background claimant or an outbound HTTP attempt. +/// +public sealed class ProposalWebhookDurabilityApiTests(HostedWorkerDisabledTestWebApplicationFactory factory) + : IClassFixture +{ + [Theory] + [InlineData(false, "archived")] + [InlineData(true, "restored")] + public async Task LifecycleProposal_StagesOnePendingDeliveryBeforeFirstPostCommitNotification( + bool initiallyArchived, + string expectedOperation) + { + var probe = new PostCommitProbe(); + using var durabilityFactory = CreateDurabilityFactory(probe); + var scenario = await SeedLifecycleProposalAsync(durabilityFactory, initiallyArchived); + probe.SetObservation(mutation => ObserveCommittedStateAsync(durabilityFactory, scenario, mutation)); + using var client = durabilityFactory.CreateClient(); + Authenticate(client, scenario); + + var response = await ExecuteAsync(client, scenario.ProposalId); + + response.EnsureSuccessStatusCode(); + probe.CommittedMutations.Should().ContainSingle(); + var mutation = probe.CommittedMutations.Single(); + mutation.BoardId.Should().Be(scenario.BoardId); + mutation.EntityType.Should().Be("card"); + mutation.Operation.Should().Be(expectedOperation); + mutation.EntityId.Should().Be(scenario.CardId); + probe.ObservationException.Should().BeNull( + "the separate observer scope must read the committed board and queue at the first post-commit callback"); + probe.FirstCommittedState.Should().NotBeNull( + "the first post-commit notification observes the real SQLite queue before any worker can claim it"); + var committed = probe.FirstCommittedState!; + committed.IsArchived.Should().Be(!initiallyArchived); + committed.ProposalStatus.Should().Be(ProposalStatus.Applied); + committed.LastTriggeredAt.Should().NotBeNull(); + committed.Deliveries.Should().ContainSingle(); + var delivery = committed.Deliveries.Single(); + delivery.Status.Should().Be(WebhookDeliveryStatus.Pending); + delivery.AttemptCount.Should().Be(0); + delivery.BoardId.Should().Be(scenario.BoardId); + delivery.EventType.Should().Be($"card.{expectedOperation}"); + delivery.Payload.Should().Contain(scenario.CardId.ToString()); + } + + [Fact] + public async Task PostCommitNotificationFailure_LeavesCommittedPendingDeliveryRecoverableFromNewScope() + { + var probe = new PostCommitProbe { ThrowAfterObservation = true }; + using var durabilityFactory = CreateDurabilityFactory(probe); + var scenario = await SeedLifecycleProposalAsync(durabilityFactory, archived: false); + probe.SetObservation(mutation => ObserveCommittedStateAsync(durabilityFactory, scenario, mutation)); + using (var client = durabilityFactory.CreateClient()) + { + Authenticate(client, scenario); + var response = await ExecuteAsync(client, scenario.ProposalId); + response.EnsureSuccessStatusCode(); + } + + probe.CommittedMutations.Should().ContainSingle( + "the controlled post-commit notification failure occurs only after the transaction is durable"); + using var recoveryScope = durabilityFactory.Services.CreateScope(); + var recoveryDatabase = recoveryScope.ServiceProvider.GetRequiredService(); + var recovered = await recoveryDatabase.OutboundWebhookDeliveries + .Where(candidate => candidate.SubscriptionId == scenario.SubscriptionId) + .ToListAsync(); + recovered.Should().ContainSingle(); + recovered.Single().Status.Should().Be(WebhookDeliveryStatus.Pending); + (await recoveryDatabase.Cards.SingleAsync(candidate => candidate.Id == scenario.CardId)).IsArchived.Should().BeTrue(); + (await recoveryDatabase.AutomationProposals.SingleAsync(candidate => candidate.Id == scenario.ProposalId)).Status + .Should().Be(ProposalStatus.Applied); + } + + [Fact] + public async Task FinalSaveFailureAfterStaging_RollsBackMutationReceiptsQueueAndTriggerMarker() + { + var probe = new PostCommitProbe(); + var failure = new StagedDeliverySaveFailureInterceptor(); + using var durabilityFactory = CreateDurabilityFactory(probe, failure); + var scenario = await SeedLifecycleProposalAsync(durabilityFactory, archived: false); + using var client = durabilityFactory.CreateClient(); + Authenticate(client, scenario); + + var response = await ExecuteAsync(client, scenario.ProposalId); + + response.IsSuccessStatusCode.Should().BeFalse(); + failure.Injected.Should().BeTrue("the failure must occur at the final save after a delivery row has been staged"); + probe.CommittedMutations.Should().BeEmpty("post-commit notification must not run after the transaction rolls back"); + using var verificationScope = durabilityFactory.Services.CreateScope(); + var database = verificationScope.ServiceProvider.GetRequiredService(); + (await database.Cards.SingleAsync(candidate => candidate.Id == scenario.CardId)).IsArchived.Should().BeFalse(); + (await database.AutomationProposals.SingleAsync(candidate => candidate.Id == scenario.ProposalId)).Status + .Should().NotBe(ProposalStatus.Applied); + (await database.AuditLogs.Where(candidate => candidate.Changes != null + && candidate.Changes.Contains(scenario.ProposalId.ToString())).ToListAsync()).Should().BeEmpty(); + (await database.OutboundWebhookDeliveries.AnyAsync(candidate => candidate.SubscriptionId == scenario.SubscriptionId)) + .Should().BeFalse(); + (await database.OutboundWebhookSubscriptions.SingleAsync(candidate => candidate.Id == scenario.SubscriptionId)).LastTriggeredAt + .Should().BeNull("rollback clears tracked staged state before failure recovery persists its own status"); + } + + [Fact] + public async Task AlreadyAppliedRetry_DoesNotStageOrDuplicateDelivery() + { + var probe = new PostCommitProbe(); + using var durabilityFactory = CreateDurabilityFactory(probe); + var scenario = await SeedLifecycleProposalAsync(durabilityFactory, archived: false); + probe.SetObservation(mutation => ObserveCommittedStateAsync(durabilityFactory, scenario, mutation)); + using var client = durabilityFactory.CreateClient(); + Authenticate(client, scenario); + + (await ExecuteAsync(client, scenario.ProposalId, "first-execution")).EnsureSuccessStatusCode(); + (await ExecuteAsync(client, scenario.ProposalId, "already-applied-retry")).EnsureSuccessStatusCode(); + + probe.StagedMutations.Should().ContainSingle(); + probe.CommittedMutations.Should().ContainSingle(); + using var verificationScope = durabilityFactory.Services.CreateScope(); + var database = verificationScope.ServiceProvider.GetRequiredService(); + (await database.OutboundWebhookDeliveries.CountAsync(candidate => candidate.SubscriptionId == scenario.SubscriptionId)) + .Should().Be(1); + } + + private static async Task SeedLifecycleProposalAsync( + WebApplicationFactory factory, + bool archived, + string eventFilter = "card.*") + { + using var client = factory.CreateClient(); + // ApiTestHarness adds its own uniqueness suffix; keep this stem within the API username limit. + var user = await ApiTestHarness.AuthenticateAsync(client, $"whd-{Guid.NewGuid():N}"); + var boardId = await ApiTestHarness.CreateBoardWithColumnAsync(client, "Durable webhook proposal"); + var board = (await client.GetFromJsonAsync($"/api/boards/{boardId}"))!; + var cardResponse = await client.PostAsJsonAsync($"/api/boards/{boardId}/cards", + new CreateCardDto(boardId, board.Columns.Single().Id, "Lifecycle target", null, null, null)); + cardResponse.EnsureSuccessStatusCode(); + var card = (await cardResponse.Content.ReadFromJsonAsync())!; + + if (archived) + { + var archive = await client.PostAsJsonAsync($"/api/boards/{boardId}/cards/{card.Id}/archive", + new CardLifecycleDto(card.UpdatedAt, null)); + archive.EnsureSuccessStatusCode(); + card = (await archive.Content.ReadFromJsonAsync())!; + } + + Guid subscriptionId; + using (var seedScope = factory.Services.CreateScope()) + { + var database = seedScope.ServiceProvider.GetRequiredService(); + var subscription = new OutboundWebhookSubscription( + boardId, + user.UserId, + "https://example.test/durable-proposal-events", + "synthetic-signing-secret", + [eventFilter]); + database.OutboundWebhookSubscriptions.Add(subscription); + await database.SaveChangesAsync(); + subscriptionId = subscription.Id; + } + + var proposalResponse = await client.PostAsJsonAsync("/api/automation/proposals", new CreateProposalDto( + ProposalSourceType.Manual, + user.UserId, + "Durable webhook lifecycle proposal", + RiskLevel.Low, + Guid.NewGuid().ToString("N"), + boardId, + Operations: + [ + new CreateProposalOperationDto( + 0, + archived ? "restore-lifecycle" : "archive-lifecycle", + "card", + System.Text.Json.JsonSerializer.Serialize(new { cardId = card.Id, expectedUpdatedAt = card.UpdatedAt }), + Guid.NewGuid().ToString("N"), + card.Id.ToString()) + ])); + proposalResponse.StatusCode.Should().Be(HttpStatusCode.Created, await proposalResponse.Content.ReadAsStringAsync()); + var proposal = (await proposalResponse.Content.ReadFromJsonAsync())!; + (await client.PostAsync($"/api/automation/proposals/{proposal.Id}/approve", null)).EnsureSuccessStatusCode(); + + return new Scenario(user.UserId, user.Token, boardId, card.Id, subscriptionId, proposal.Id, card.UpdatedAt); + } + + private static void Authenticate(HttpClient client, Scenario scenario) => + client.DefaultRequestHeaders.Authorization = new AuthenticationHeaderValue("Bearer", scenario.Token); + + private static async Task ExecuteAsync(HttpClient client, Guid proposalId, string? idempotencyKey = null) + { + using var request = new HttpRequestMessage(HttpMethod.Post, $"/api/automation/proposals/{proposalId}/execute"); + request.Headers.Add("Idempotency-Key", idempotencyKey ?? Guid.NewGuid().ToString("N")); + return await client.SendAsync(request); + } + + private static async Task ObserveCommittedStateAsync( + WebApplicationFactory factory, + Scenario scenario, + BoardRealtimeEvent mutation) + { + using var observerScope = factory.Services.CreateScope(); + var database = observerScope.ServiceProvider.GetRequiredService(); + var card = await database.Cards.SingleAsync(candidate => candidate.Id == scenario.CardId); + var proposal = await database.AutomationProposals.SingleAsync(candidate => candidate.Id == scenario.ProposalId); + var subscription = await database.OutboundWebhookSubscriptions.SingleAsync(candidate => candidate.Id == scenario.SubscriptionId); + var deliveries = await database.OutboundWebhookDeliveries + .Where(candidate => candidate.SubscriptionId == scenario.SubscriptionId) + .ToListAsync(); + mutation.EntityId.Should().Be(scenario.CardId); + return new CommittedState(card.IsArchived, proposal.Status, subscription.LastTriggeredAt, deliveries); + } + + private WebApplicationFactory CreateDurabilityFactory( + PostCommitProbe probe, + SaveChangesInterceptor? saveChangesInterceptor = null) => + factory.WithWebHostBuilder(builder => builder.ConfigureTestServices(services => + { + if (saveChangesInterceptor is not null) + { + services.RemoveAll>(); + services.RemoveAll(); + services.AddDbContext((provider, options) => + { + var configuration = provider.GetRequiredService(); + options.UseTaskdeckSqlite( + configuration.GetConnectionString("DefaultConnection")!, + configuration.GetSection("Database").Get() ?? new DatabaseSettings()) + .AddInterceptors(saveChangesInterceptor); + }); + } + + services.RemoveAll(); + services.AddScoped(provider => new ProbeTransactionalNotifier( + provider.GetRequiredService(), probe)); + })); + + private sealed record Scenario( + Guid UserId, + string Token, + Guid BoardId, + Guid CardId, + Guid SubscriptionId, + Guid ProposalId, + DateTimeOffset CardUpdatedAt); + + private sealed record CommittedState( + bool IsArchived, + ProposalStatus ProposalStatus, + DateTimeOffset? LastTriggeredAt, + IReadOnlyList Deliveries); + + private sealed class ProbeTransactionalNotifier( + IOutboundWebhookService webhooks, + PostCommitProbe probe) : IBoardRealtimeNotifier, ITransactionalBoardMutationNotifier + { + public Task NotifyBoardMutationAsync(BoardRealtimeEvent mutation, CancellationToken cancellationToken = default) + { + probe.LegacyMutations.Add(mutation); + return Task.CompletedTask; + } + + public async Task StageBoardMutationAsync(BoardRealtimeEvent mutation, CancellationToken cancellationToken = default) + { + probe.StagedMutations.Add(mutation); + var result = await webhooks.StageBoardMutationAsync(mutation, cancellationToken); + result.IsSuccess.Should().BeTrue(result.ErrorMessage); + } + + public Task NotifyCommittedBoardMutationAsync( + BoardRealtimeEvent mutation, + CancellationToken cancellationToken = default) => probe.ObserveCommittedAsync(mutation); + } + + private sealed class PostCommitProbe + { + private Func>? _observation; + + public List StagedMutations { get; } = []; + public List CommittedMutations { get; } = []; + public List LegacyMutations { get; } = []; + public CommittedState? FirstCommittedState { get; private set; } + public Exception? ObservationException { get; private set; } + public bool ThrowAfterObservation { get; init; } + + public void SetObservation(Func> observation) => _observation = observation; + + public async Task ObserveCommittedAsync(BoardRealtimeEvent mutation) + { + CommittedMutations.Add(mutation); + if (_observation is not null && FirstCommittedState is null) + { + try + { + FirstCommittedState = await _observation(mutation); + } + catch (Exception ex) + { + ObservationException = ex; + throw; + } + } + if (ThrowAfterObservation) + throw new InvalidOperationException("Controlled post-commit notification failure."); + } + } + + private sealed class StagedDeliverySaveFailureInterceptor : SaveChangesInterceptor + { + public bool Injected { get; private set; } + + public override ValueTask> SavingChangesAsync( + DbContextEventData eventData, + InterceptionResult result, + CancellationToken cancellationToken = default) + { + if (!Injected && eventData.Context?.ChangeTracker.Entries() + .Any(entry => entry.State == EntityState.Added) == true) + { + Injected = true; + throw new DbUpdateException("Controlled final save failure after webhook staging."); + } + + return ValueTask.FromResult(result); + } + } +} diff --git a/backend/tests/Taskdeck.Application.Tests/Services/DeferredBoardRealtimeNotifierTests.cs b/backend/tests/Taskdeck.Application.Tests/Services/DeferredBoardRealtimeNotifierTests.cs index 7caf9150fb..9e33ac2fd0 100644 --- a/backend/tests/Taskdeck.Application.Tests/Services/DeferredBoardRealtimeNotifierTests.cs +++ b/backend/tests/Taskdeck.Application.Tests/Services/DeferredBoardRealtimeNotifierTests.cs @@ -84,6 +84,49 @@ public async Task Notify_ShouldNotThrow_WhenNoDownstreamNotifierIsConfigured() deferred.PendingCount.Should().Be(0); } + [Fact] + public async Task Prepare_ShouldStageWithoutDraining_ThenFlushCommittedWithoutDoubleEnqueue() + { + var inner = new TransactionalNotifier(); + var deferred = new DeferredBoardRealtimeNotifier(inner); + var mutation = Event("updated"); + + await deferred.NotifyBoardMutationAsync(mutation); + await deferred.PrepareAsync(); + await deferred.PrepareAsync(); + + inner.Staged.Should().ContainSingle().Which.Should().BeSameAs(mutation); + inner.Published.Should().BeEmpty(); + inner.LegacyPublished.Should().BeEmpty(); + deferred.PendingCount.Should().Be(1); + deferred.PreparedCount.Should().Be(1); + + await deferred.FlushAsync(); + await deferred.FlushAsync(); + + inner.Published.Should().ContainSingle().Which.Should().BeSameAs(mutation); + inner.Staged.Should().ContainSingle(); + inner.LegacyPublished.Should().BeEmpty(); + deferred.PendingCount.Should().Be(0); + deferred.PreparedCount.Should().Be(0); + } + + [Fact] + public async Task Prepare_ShouldLeaveFailedEventPendingWithoutMarkingItPrepared() + { + var inner = new TransactionalNotifier { StageException = new InvalidOperationException("stage failed") }; + var deferred = new DeferredBoardRealtimeNotifier(inner); + await deferred.NotifyBoardMutationAsync(Event("updated")); + + var act = () => deferred.PrepareAsync(); + + await act.Should().ThrowAsync().WithMessage("stage failed"); + deferred.PendingCount.Should().Be(1); + deferred.PreparedCount.Should().Be(0); + inner.Published.Should().BeEmpty(); + inner.LegacyPublished.Should().BeEmpty(); + } + private static BoardRealtimeEvent Event(string operation) => new(Guid.NewGuid(), "card", operation, Guid.NewGuid(), DateTimeOffset.UtcNow); @@ -108,4 +151,33 @@ public Task NotifyBoardMutationAsync(BoardRealtimeEvent mutation, CancellationTo throw new InvalidOperationException("channel down"); } } + + private sealed class TransactionalNotifier : IBoardRealtimeNotifier, ITransactionalBoardMutationNotifier + { + public List Staged { get; } = []; + public List Published { get; } = []; + public List LegacyPublished { get; } = []; + public Exception? StageException { get; init; } + + public Task NotifyBoardMutationAsync(BoardRealtimeEvent mutation, CancellationToken cancellationToken = default) + { + LegacyPublished.Add(mutation); + return Task.CompletedTask; + } + + public Task StageBoardMutationAsync(BoardRealtimeEvent mutation, CancellationToken cancellationToken = default) + { + if (StageException is not null) + throw StageException; + + Staged.Add(mutation); + return Task.CompletedTask; + } + + public Task NotifyCommittedBoardMutationAsync(BoardRealtimeEvent mutation, CancellationToken cancellationToken = default) + { + Published.Add(mutation); + return Task.CompletedTask; + } + } } diff --git a/backend/tests/Taskdeck.Application.Tests/Services/OutboundWebhookServiceTests.cs b/backend/tests/Taskdeck.Application.Tests/Services/OutboundWebhookServiceTests.cs index 6295eb62b1..c338779255 100644 --- a/backend/tests/Taskdeck.Application.Tests/Services/OutboundWebhookServiceTests.cs +++ b/backend/tests/Taskdeck.Application.Tests/Services/OutboundWebhookServiceTests.cs @@ -6,6 +6,7 @@ using Taskdeck.Application.Services; using Taskdeck.Domain.Common; using Taskdeck.Domain.Entities; +using Taskdeck.Domain.Enums; using Taskdeck.Domain.Exceptions; using Xunit; @@ -325,6 +326,77 @@ public async Task EnqueueBoardMutationAsync_ShouldOnlyQueueDeliveriesForMatching payload.RootElement.TryGetProperty("boardId", out _).Should().BeTrue(); payload.RootElement.TryGetProperty("DeliveryId", out _).Should().BeFalse(); payload.RootElement.GetProperty("deliveryId").GetGuid().Should().Be(createdDeliveries[0].Id); + _unitOfWorkMock.Verify( + unitOfWork => unitOfWork.SaveChangesAsync(It.IsAny()), + Times.Once); + } + + [Fact] + public async Task StageBoardMutationAsync_ShouldPrepareMatchingPendingDeliveriesWithoutSaving() + { + var boardId = Guid.NewGuid(); + var matching = new OutboundWebhookSubscription( + boardId, + Guid.NewGuid(), + "https://example.com/matching", + "secret", + ["card.*"]); + var nonMatching = new OutboundWebhookSubscription( + boardId, + Guid.NewGuid(), + "https://example.com/non-matching", + "secret", + ["proposal.*"]); + _subscriptionRepositoryMock + .Setup(repository => repository.GetActiveByBoardAsync(boardId, It.IsAny())) + .ReturnsAsync([matching, nonMatching]); + + var createdDeliveries = new List(); + _deliveryRepositoryMock + .Setup(repository => repository.AddAsync(It.IsAny(), It.IsAny())) + .Callback((delivery, _) => createdDeliveries.Add(delivery)) + .ReturnsAsync((OutboundWebhookDelivery delivery, CancellationToken _) => delivery); + + var service = new OutboundWebhookService(_unitOfWorkMock.Object); + var result = await service.StageBoardMutationAsync( + new BoardRealtimeEvent(boardId, "card", "updated", Guid.NewGuid(), DateTimeOffset.UtcNow)); + + result.IsSuccess.Should().BeTrue(); + createdDeliveries.Should().ContainSingle(); + createdDeliveries[0].SubscriptionId.Should().Be(matching.Id); + createdDeliveries[0].Status.Should().Be(WebhookDeliveryStatus.Pending); + matching.LastTriggeredAt.Should().NotBeNull(); + nonMatching.LastTriggeredAt.Should().BeNull(); + _unitOfWorkMock.Verify( + unitOfWork => unitOfWork.SaveChangesAsync(It.IsAny()), + Times.Never); + } + + [Fact] + public async Task StageBoardMutationAsync_ShouldPropagatePreparationFailureWithoutSaving() + { + var boardId = Guid.NewGuid(); + var subscription = new OutboundWebhookSubscription( + boardId, + Guid.NewGuid(), + "https://example.com/hook", + "secret"); + _subscriptionRepositoryMock + .Setup(repository => repository.GetActiveByBoardAsync(boardId, It.IsAny())) + .ReturnsAsync([subscription]); + _deliveryRepositoryMock + .Setup(repository => repository.AddAsync(It.IsAny(), It.IsAny())) + .ThrowsAsync(new InvalidOperationException("staging failed")); + + var service = new OutboundWebhookService(_unitOfWorkMock.Object); + + var act = () => service.StageBoardMutationAsync( + new BoardRealtimeEvent(boardId, "card", "updated", Guid.NewGuid(), DateTimeOffset.UtcNow)); + + await act.Should().ThrowAsync().WithMessage("staging failed"); + _unitOfWorkMock.Verify( + unitOfWork => unitOfWork.SaveChangesAsync(It.IsAny()), + Times.Never); } [Fact] diff --git a/backend/tests/Taskdeck.Application.Tests/Services/ProposalLifecycleNotificationTests.cs b/backend/tests/Taskdeck.Application.Tests/Services/ProposalLifecycleNotificationTests.cs index 6337f0a288..d71c690594 100644 --- a/backend/tests/Taskdeck.Application.Tests/Services/ProposalLifecycleNotificationTests.cs +++ b/backend/tests/Taskdeck.Application.Tests/Services/ProposalLifecycleNotificationTests.cs @@ -7,6 +7,7 @@ using Taskdeck.Domain.Common; using Taskdeck.Domain.Entities; using Taskdeck.Domain.Enums; +using Taskdeck.Domain.Exceptions; using Xunit; namespace Taskdeck.Application.Tests.Services; @@ -103,6 +104,8 @@ public async Task ExecuteProposal_ShouldPublishNoLifecycleNotification_WhenALate _unitOfWorkMock.Verify(u => u.SaveChangesAsync(It.IsAny()), Times.AtLeastOnce); _rolledBack.Should().BeTrue("the failing operation must roll the archive back"); _committed.Should().BeFalse(); + _notifier.Staged.Should().BeEmpty( + "durable delivery preparation waits until every operation succeeds"); _notifier.Published.Should().BeEmpty( "a lifecycle change that was rolled back must never reach realtime subscribers"); } @@ -145,6 +148,8 @@ public async Task ExecuteProposal_ShouldDeferTheDetachedChildEventsTooAndPublish "every event staged inside the transaction waits for the commit"); _notifier.Published.Select(p => (p.Mutation.Operation, p.Mutation.EntityId)) .Should().Equal(("archived", (Guid?)card.Id), ("updated", child.Id)); + _notifier.Staged.Select(e => (e.Operation, e.EntityId)) + .Should().Equal(("archived", (Guid?)card.Id), ("updated", child.Id)); } [Fact] @@ -204,6 +209,93 @@ public async Task ExecuteProposal_ShouldPublishOneLifecycleNotificationAfterComm published.Mutation.EntityId.Should().Be(card.Id); published.CommittedAtPublishTime.Should().BeTrue( "the event must be published after the proposal transaction commits, not before"); + _notifier.Staged.Should().ContainSingle().Which.Should().BeSameAs(published.Mutation); + _notifier.CommittedAtStageTime.Should().ContainSingle().Which.Should().BeFalse( + "durable webhook rows must be prepared before the transaction commits"); + } + + [Fact] + public async Task ExecuteProposal_ShouldRollbackAndPublishNothing_WhenDurablePreparationFails() + { + var failingNotifier = new FailingTransactionalNotifier(); + var executor = new AutomationExecutorService( + _unitOfWorkMock.Object, + _proposalServiceMock.Object, + _policyEngineMock.Object, + new CardService(_unitOfWorkMock.Object, failingNotifier), + new BoardService(_unitOfWorkMock.Object), + new ColumnService(_unitOfWorkMock.Object), + logger: null, + assignments: null, + realtimeNotifier: failingNotifier); + var (board, column, card) = SeedBoard(archived: false); + var proposalId = Guid.NewGuid(); + var entity = ArrangeApprovedProposal( + proposalId, + board.Id, + [LifecycleOperation(proposalId, sequence: 0, card, archive: true)]); + + var result = await executor.ExecuteProposalAsync(proposalId, "execution-key"); + + result.IsSuccess.Should().BeFalse(); + result.ErrorCode.Should().Be(ErrorCodes.UnexpectedError); + failingNotifier.StageAttempts.Should().Be(1); + failingNotifier.CommittedAttempts.Should().Be(0); + failingNotifier.LegacyAttempts.Should().Be(0); + _rolledBack.Should().BeTrue(); + _committed.Should().BeFalse(); + entity.Status.Should().Be(ProposalStatus.Failed); + } + + [Fact] + public async Task ExecuteProposal_ShouldCommitWithoutStagingOrPublishing_WhenNoBufferedMutationExists() + { + var board = TestDataBuilder.CreateBoard(); + _boardRepoMock.Setup(r => r.GetByIdAsync(board.Id, It.IsAny())).ReturnsAsync(board); + var proposalId = Guid.NewGuid(); + ArrangeApprovedProposal(proposalId, board.Id, []); + + var result = await _executor.ExecuteProposalAsync(proposalId, "execution-key"); + + result.IsSuccess.Should().BeTrue(); + _committed.Should().BeTrue(); + _notifier.Staged.Should().BeEmpty(); + _notifier.Published.Should().BeEmpty(); + } + + [Fact] + public async Task ExecuteProposal_ShouldNotStageOrPublish_WhenProposalWasAlreadyApplied() + { + var proposalId = Guid.NewGuid(); + var proposal = new ProposalDto( + proposalId, + ProposalSourceType.Manual, + null, + Guid.NewGuid(), + Guid.NewGuid(), + ProposalStatus.Applied, + RiskLevel.Low, + "Already applied", + null, + null, + DateTimeOffset.UtcNow, + DateTimeOffset.UtcNow, + DateTime.UtcNow.AddDays(1), + DateTime.UtcNow, + Guid.NewGuid(), + DateTime.UtcNow, + null, + $"corr-{proposalId:N}", + []); + _proposalServiceMock.Setup(s => s.GetProposalByIdAsync(proposalId, It.IsAny())) + .ReturnsAsync(Result.Success(proposal)); + + var result = await _executor.ExecuteProposalAsync(proposalId, "execution-key"); + + result.IsSuccess.Should().BeTrue(); + _unitOfWorkMock.Verify(u => u.BeginTransactionAsync(It.IsAny()), Times.Never); + _notifier.Staged.Should().BeEmpty(); + _notifier.Published.Should().BeEmpty(); } [Theory] @@ -337,7 +429,7 @@ private AutomationProposal ArrangeApprovedProposal(Guid proposalId, Guid boardId /// Captures each published event together with whether the unit of work had already committed /// at that moment — the ordering assertion the issue asks for. /// - private sealed class RecordingBoardRealtimeNotifier : IBoardRealtimeNotifier + private sealed class RecordingBoardRealtimeNotifier : IBoardRealtimeNotifier, ITransactionalBoardMutationNotifier { private readonly Func _committedProbe; private readonly List<(BoardRealtimeEvent Mutation, bool CommittedAtPublishTime)> _published = new(); @@ -345,12 +437,52 @@ private sealed class RecordingBoardRealtimeNotifier : IBoardRealtimeNotifier public RecordingBoardRealtimeNotifier(Func committedProbe) => _committedProbe = committedProbe; public IReadOnlyList<(BoardRealtimeEvent Mutation, bool CommittedAtPublishTime)> Published => _published; + public List Staged { get; } = []; + public List CommittedAtStageTime { get; } = []; public Task NotifyBoardMutationAsync(BoardRealtimeEvent mutation, CancellationToken cancellationToken = default) { _published.Add((mutation, _committedProbe())); return Task.CompletedTask; } + + public Task StageBoardMutationAsync(BoardRealtimeEvent mutation, CancellationToken cancellationToken = default) + { + Staged.Add(mutation); + CommittedAtStageTime.Add(_committedProbe()); + return Task.CompletedTask; + } + + public Task NotifyCommittedBoardMutationAsync(BoardRealtimeEvent mutation, CancellationToken cancellationToken = default) + { + _published.Add((mutation, _committedProbe())); + return Task.CompletedTask; + } + } + + private sealed class FailingTransactionalNotifier : IBoardRealtimeNotifier, ITransactionalBoardMutationNotifier + { + public int LegacyAttempts { get; private set; } + public int StageAttempts { get; private set; } + public int CommittedAttempts { get; private set; } + + public Task NotifyBoardMutationAsync(BoardRealtimeEvent mutation, CancellationToken cancellationToken = default) + { + LegacyAttempts++; + return Task.CompletedTask; + } + + public Task StageBoardMutationAsync(BoardRealtimeEvent mutation, CancellationToken cancellationToken = default) + { + StageAttempts++; + throw new InvalidOperationException("durable staging failed"); + } + + public Task NotifyCommittedBoardMutationAsync(BoardRealtimeEvent mutation, CancellationToken cancellationToken = default) + { + CommittedAttempts++; + return Task.CompletedTask; + } } /// A notification channel that is simply down — the shape the executor must survive. diff --git a/docs/IMPLEMENTATION_MASTERPLAN.md b/docs/IMPLEMENTATION_MASTERPLAN.md index f456f2f929..36544c30f9 100644 --- a/docs/IMPLEMENTATION_MASTERPLAN.md +++ b/docs/IMPLEMENTATION_MASTERPLAN.md @@ -2,6 +2,26 @@ Last Updated: 2026-09-12 +## Proposal webhook durability candidate (2026-09-12, #3024) + +Reuse the existing outbound delivery queue for events already buffered by the proposal executor. +Prepare delivery rows before the final transaction save, commit them with the proposal effects, +then publish realtime after commit without enqueueing webhooks again. The existing queue worker +provides later delivery; no new schema, sweep or worker policy is needed. Preparation failure must +roll back the proposal. This reduces silent integration gaps while retaining review-first writes. + +The implementation, 49 focused Application/Composite API tests, five real SQLite visibility/rollback/ +lost-flush tests and one independent review are complete. Two lifecycle controls fail when durable +preparation is omitted and pass with the reviewed bytes restored. The complete backend gate now +passes 9,802 tests with 34 existing skips and no failures. Complete exact-head hosted qualification +before closing #3024. Broader notification +producer changes, live HTTP and release qualification remain outside this slice. + +The child now includes the parent's corrected chat ID/revision contracts. Only those nine backend +tool/registration/test files differ from the full-backend checkpoint; their 91 Application and one +API registry cases pass on the parent. Webhook source and durability tests remain unchanged. +Qualify the combined head and retarget after parent delivery before merging this child. + ## Typed relations candidate (2026-09-12, #2092) The candidate includes delivered archive recovery #3059 and draft correction #3064 through diff --git a/docs/STATUS.md b/docs/STATUS.md index 369220b01b..10891c0325 100644 --- a/docs/STATUS.md +++ b/docs/STATUS.md @@ -2,6 +2,26 @@ Last Updated: 2026-09-12 +Proposal webhook durability (#3024) is implemented on a candidate branch. Events already buffered +by the proposal executor now prepare filtered `Pending` delivery rows in its existing transaction, +before the Applied-status save. Delivery rows, subscription trigger timestamps and proposal effects +commit together or roll back together. Post-commit notification sends only the best-effort realtime +channel; the existing delivery worker can claim committed webhook rows even if that flush is lost. +This closes the missing-delivery window without changing review, approval or Apply requirements. +No new queue schema or retry policy is introduced. Immediate notification producers and the separate +assignment collector retain their current behavior; this is not an account-wide outbox conversion. + +Focused Application tests pass 39 cases and Composite API tests pass 10. Five real SQLite API +tests pass for first-notification queue visibility, rollback, lost post-commit callback recovery and +already-applied deduplication. Omitting durable preparation makes both lifecycle controls fail; +restoring the reviewed executor bytes makes all five pass. Independent source review is clean. +The complete backend gate passes 9,802 tests across all six projects with 34 existing skips and +zero failures at `0a65b536b`. The later inherited chat correction changes only its nine tool, +registration and test files; those changes pass 91 Application cases and one API registry case +on the parent. Webhook implementation and durability tests are unchanged. Exact-head hosted +qualification remains required for the combined tree. No actual process-kill, external HTTP +delivery or release acceptance is claimed. + The typed-relation candidate includes delivered archive recovery #3059 and draft settlement #3064 through main `9a8c14c6b`. Their required hosted gates passed; #3033 is closed and #3023 retains its broader residuals. Integration resolved only concurrent STATUS/MASTERPLAN records, preserving