From fb71344be21e8f64296011a65e3bbf3ee7b1d878 Mon Sep 17 00:00:00 2001 From: SzymonPobiega Date: Mon, 19 Oct 2026 14:37:55 +0200 Subject: [PATCH] Spike managing saga timeouts. Only for RavenDB persistence for now --- docs/saga-timeout-management-design.md | 455 ++++++++++++++++ .../When_a_saga_timeout_is_requested.cs | 89 ++++ .../Auth/MessageAction.cs | 5 +- .../Auth/Permissions.cs | 5 + .../Auth/RolePermissions.cs | 2 + .../Abstractions/BasePersistence.cs | 2 + .../Implementation/SagaTimeoutDataStore.cs | 26 + .../ExpirationManager.cs | 13 + .../RavenPersistence.cs | 3 + .../SagaTimeouts/SagaTimeout.cs | 87 ++++ .../SagaTimeouts/SagaTimeoutDataStore.cs | 485 ++++++++++++++++++ .../SagaTimeouts/SagaTimeoutsIndex.cs | 43 ++ ...ontrol.Persistence.Tests.PostgreSql.csproj | 3 + .../RavenPersistedTypes.Verify.approved.txt | 2 + ...Control.Persistence.Tests.SqlServer.csproj | 4 +- .../SagaTimeouts/SagaTimeoutDataStoreTests.cs | 275 ++++++++++ .../SagaTimeouts/ISagaTimeoutDataStore.cs | 47 ++ .../SagaTimeouts/SagaTimeoutModel.cs | 155 ++++++ .../SagaTimeouts/SagaTimeoutView.cs | 90 ++++ .../APIApprovals.HttpApiRoutes.approved.txt | 8 + .../SagaTimeoutManagementRequestTests.cs | 96 ++++ .../Infrastructure/NServiceBusFactory.cs | 1 + .../SagaTimeouts/RequestSagaTimeoutChange.cs | 21 + .../RequestSagaTimeoutChangeHandler.cs | 89 ++++ .../SagaTimeoutPlatformMessages.cs | 72 +++ .../SagaTimeouts/SagaTimeoutReportHandler.cs | 47 ++ .../SagaTimeoutReportProcessor.cs | 70 +++ .../SagaTimeouts/SagaTimeoutsComponent.cs | 15 + .../Web/SagaTimeoutsController.cs | 219 ++++++++ .../ServiceControlMainInstance.cs | 2 + 30 files changed, 2429 insertions(+), 2 deletions(-) create mode 100644 docs/saga-timeout-management-design.md create mode 100644 src/ServiceControl.AcceptanceTests.RavenDB/SagaTimeouts/When_a_saga_timeout_is_requested.cs create mode 100644 src/ServiceControl.Persistence.EFCore/Implementation/SagaTimeoutDataStore.cs create mode 100644 src/ServiceControl.Persistence.RavenDB/SagaTimeouts/SagaTimeout.cs create mode 100644 src/ServiceControl.Persistence.RavenDB/SagaTimeouts/SagaTimeoutDataStore.cs create mode 100644 src/ServiceControl.Persistence.RavenDB/SagaTimeouts/SagaTimeoutsIndex.cs create mode 100644 src/ServiceControl.Persistence.Tests/SagaTimeouts/SagaTimeoutDataStoreTests.cs create mode 100644 src/ServiceControl.Persistence/SagaTimeouts/ISagaTimeoutDataStore.cs create mode 100644 src/ServiceControl.Persistence/SagaTimeouts/SagaTimeoutModel.cs create mode 100644 src/ServiceControl.Persistence/SagaTimeouts/SagaTimeoutView.cs create mode 100644 src/ServiceControl.UnitTests/SagaTimeouts/SagaTimeoutManagementRequestTests.cs create mode 100644 src/ServiceControl/SagaTimeouts/RequestSagaTimeoutChange.cs create mode 100644 src/ServiceControl/SagaTimeouts/RequestSagaTimeoutChangeHandler.cs create mode 100644 src/ServiceControl/SagaTimeouts/SagaTimeoutPlatformMessages.cs create mode 100644 src/ServiceControl/SagaTimeouts/SagaTimeoutReportHandler.cs create mode 100644 src/ServiceControl/SagaTimeouts/SagaTimeoutReportProcessor.cs create mode 100644 src/ServiceControl/SagaTimeouts/SagaTimeoutsComponent.cs create mode 100644 src/ServiceControl/SagaTimeouts/Web/SagaTimeoutsController.cs diff --git a/docs/saga-timeout-management-design.md b/docs/saga-timeout-management-design.md new file mode 100644 index 0000000000..2c546d0a44 --- /dev/null +++ b/docs/saga-timeout-management-design.md @@ -0,0 +1,455 @@ +# Saga timeout management design + +Status: proposal. The endpoint side exists on the NServiceBus `saga-timeout-full-messages` branch. ServiceControl and ServicePulse do not implement any of this yet. + +## What it is + +Saga timeout management lets an operator see every pending saga timeout in the system, see how each timeout changed over time, and reschedule, trigger, or cancel a timeout from ServicePulse. + +The design rests on one rule: the endpoint owns the timeout, and ServiceControl keeps a projection of it. The saga persister stores the timeouts with the saga data. Every committed change is reported to the ServiceControl primary instance as a message. ServiceControl never changes a timeout directly. An operator action is a request sent to the endpoint, which applies it to the saga and reports the result through the same messages as any other change. + +This document covers: + +- the endpoint side: message contracts and the management satellite +- how the primary instance ingests and orders the messages +- the persistence model for RavenDB, SQL Server, and PostgreSQL +- the HTTP API and operator actions +- the ServicePulse pages, including the saga diagram additions +- rollout, compatibility, and open questions + +## Endpoint side + +### Timeout storage and dispatch + +A saga persister opts in by implementing `ISupportTimeoutManagement`. The saga API is `RequestTimeout`, `FindTimeouts`, `RescheduleTimeout`, and `CancelTimeout`. The persister stores a list of `StoredSagaTimeout` with the saga data: + +| Field | Meaning | +|---|---| +| `TimeoutId` | Created when the timeout is requested. It never changes. | +| `TriggerId` | Identifies the current scheduled instance. It is regenerated on every reschedule. | +| `Version` | Starts at 1. It increments on every reschedule. Cancel does not increment it. | +| `DueTime` | The last requested due time. | +| `MessageType` | Full name of the timeout message type. | +| `ContentType`, `Payload` | The serialized timeout message. | + +After the saga is saved, the endpoint sends each new or rescheduled timeout as a normal delayed message to its own input queue. The `NServiceBus.SagaTimeoutId` header carries `{TimeoutId}-{TriggerId}`. When the message arrives, the endpoint fires the timeout only if a stored timeout matches that identity. Otherwise the message is stale and is dropped without invoking the handler. Firing removes the timeout from storage in the same save as the handler's saga changes. + +### Reporting + +Reporting is opt-in: `endpointConfiguration.ReportSagaTimeoutsToServiceControl()`. It needs a persister that implements `ISupportTimeoutManagement`. The reports go to the primary instance queue. The default queue is `Particular.ServiceControl`; `ServiceControlErrorInstanceQueue(...)` overrides it. + +| Message | Sent when | Extra fields | +|---|---|---| +| `SagaTimeoutRequested` | The saga called `RequestTimeout` | `TimeoutId`, `TimeoutType`, `Version`, `TriggerId`, `DueTime`, `PayloadContentType`, `Payload` | +| `SagaTimeoutRescheduled` | The saga or an operator rescheduled a stored timeout | Same as `SagaTimeoutRequested`, plus `OperationId` | +| `SagaTimeoutCanceled` | The saga or an operator canceled a stored timeout | `TimeoutId`, `TimeoutType`, `Version`, `OperationId` | +| `SagaTimeoutTriggered` | A stored timeout fired and the handler succeeded | `TimeoutId`, `TimeoutType`, `Version`, `TriggerId` | +| `SagaTimeoutChangeRejected` | A reschedule or cancel could not be applied | `TimeoutId`, `RequestedChange`, `RequestedDueTime`, `Reason`, `OperationId` | +| `SagaCompleted` | A saga that had stored timeouts when it was loaded completed | none | + +Every message also carries: + +- `EndpointName`, `SagaType`, `SagaId`, and `CorrelationPropertyValue` +- `ChangedAt`, from the clock of the machine that made the change +- `Host` and `HostId`, the same values that heartbeats and custom checks report +- `ManagementAddress`, the transport address of the endpoint's management satellite + +`OperationId` is set only for changes an operator requested. `Reason` is `TimeoutNotFound`, `VersionMismatch`, or `SagaNotFound`. + +A completing saga reports `SagaTimeoutTriggered`, if a timeout fired, and `SagaCompleted`. It does not report the individual changes the completing handler made, because completion discards all of the saga's timeouts. A new saga that requests timeouts and completes in the same message reports nothing. + +Reports from a saga handler are pending transport operations of the incoming message. A rolled-back handler reports nothing, and with the outbox the reports are dispatched exactly when the saga changes are. The transport headers are `NServiceBus.EnclosedMessageTypes` (the full type name, for example `NServiceBus.SagaTimeoutRequested`), `NServiceBus.ContentType: application/json`, and `NServiceBus.MessageIntent: Send`. The body is System.Text.Json with PascalCase property names, and `Payload` is base64. + +### Management satellite + +When reporting is enabled, the endpoint also runs a satellite named "Saga timeout management" on the queue `{endpoint queue}.SagaTimeouts`. It applies changes that an operator requests through ServiceControl. Its address is the `ManagementAddress` in every report. + +A satellite bypasses the message pipeline, so the request is not deserialized. Every field travels in a header: + +| Header | Value | +|---|---| +| `NServiceBus.SagaTimeoutManagement.OperationId` | GUID that ServiceControl assigns to the request | +| `NServiceBus.SagaTimeoutManagement.SagaType` | Saga type full name, as reported | +| `NServiceBus.SagaTimeoutManagement.SagaId` | Saga id | +| `NServiceBus.SagaTimeoutManagement.TimeoutId` | Timeout id | +| `NServiceBus.SagaTimeoutManagement.ExpectedVersion` | The `Version` the operator saw | +| `NServiceBus.SagaTimeoutManagement.Change` | `Reschedule` or `Cancel` | +| `NServiceBus.SagaTimeoutManagement.DueTime` | New due time in the NServiceBus wire format; required for `Reschedule` | +| `NServiceBus.ContentType` | Content type of the body; required when the body is not empty | + +A `Reschedule` body, when present, replaces the timeout payload as is. An empty body keeps the stored payload. A `Cancel` must have an empty body. + +The satellite uses the outbox the same way the receive pipeline does (`TransportReceiveToPhysicalMessageConnector`). For each request, it: + +1. looks up the request's message id in the outbox; if a record exists, it dispatches the stored operations and skips to step 7 +2. begins an outbox transaction and opens a synchronized storage session on it, or on the receive transaction when the outbox is disabled +3. finds the saga metadata by type name and loads the saga by id +4. compares `ExpectedVersion` with the stored timeout +5. applies the change, saves the saga, and completes the session +6. stores the rescheduled timeout message and the reports in the outbox record, commits, and dispatches them +7. marks the outbox record as dispatched + +A missing saga, a missing timeout, or a different version produces `SagaTimeoutChangeRejected` and saves nothing. The rejection is still stored in the outbox record, so a duplicate request repeats the same rejection. A rescheduled timeout whose due time has passed is dispatched without a delay. "Trigger now" is a reschedule to the current time, so the saga handles it exactly like a timeout that came due. + +A malformed request moves to the error queue without retries. Other failures, such as a persister concurrency conflict, use the default recoverability policy. + +Without the outbox, the satellite has a gap. Suppose the saga save commits and the dispatch then fails. The retried request carries the old `ExpectedVersion`, so it is rejected with `VersionMismatch`. The change is saved but not reported, and the rescheduled timeout message is lost. ServiceControl shows the timeout as overdue, and the operator can reschedule it again. With the outbox, the retry finds the outbox record and dispatches the original operations. + +### Delivery guarantees ServiceControl must handle + +- Duplicates happen. Outbox replays, transport redelivery, and a retry after a commit whose dispatch failed all resend a report. No message carries a deduplication id. +- Reorder happens. The primary processes its input queue concurrently, and transports do not guarantee order. +- Loss happens without the outbox. Suppose the saga save commits and the dispatch fails. The retry then sees the timeout message as stale, so `SagaTimeoutTriggered` is never sent. +- A payload change without a reschedule is neither stored nor reported. A payload-only change is a reschedule with the same due time. It produces `SagaTimeoutRescheduled` with a new `Version` and `TriggerId`. + +## Ingestion in the primary instance + +### Message types and handlers + +The primary declares local copies of the six contracts in `namespace NServiceBus` so that `NServiceBus.EnclosedMessageTypes` resolves by full name. The `EndpointUsageReport` contract in `Particular.LicensingComponent` sets the precedent. A new `SagaTimeoutMessagesSerializationContext` is added to the `TypeInfoResolverChain` in `NServiceBusFactory`. + +A new `SagaTimeoutsComponent` is added to `ServiceControlMainInstance`. It registers one handler per message type, the data store, the event log mappings, and the API. Each handler converts its message into a `SagaTimeoutChange` record and calls one data-store method. The receive time comes from the injected `TimeProvider` and is stored as `ReceivedAt` beside the endpoint's `ChangedAt`. + +The handlers do not run when the primary runs with `--error-ingestion-only`, because that mode does not host the NServiceBus endpoint. Another host of the same instance must process the input queue. + +### Applying a change + +Messages arrive in any order and more than once. The store therefore treats a timeout as a state that changes only forward. The order is defined by `(Version, Rank)`: + +| Change | Status after | Rank | +|---|---|---| +| `Requested` | `Pending` | 0 | +| `Rescheduled` | `Pending` | 0 | +| `Triggered` | `Triggered` | 1 | +| `Canceled` | `Canceled` | 1 | + +A change replaces the current state only if its `(Version, Rank)` is greater than the stored one. Otherwise only its history entry is recorded. Some consequences: + +- A duplicate has the same key and changes nothing. +- `Canceled` at version 2 before `Rescheduled` at version 2 leaves the timeout `Canceled`. The reschedule is still recorded in the history. +- `Requested` at version 1 after `Rescheduled` at version 3 leaves version 3 current. +- The first change seen for a `TimeoutId` creates the timeout, whatever its kind. A `Canceled` without a prior `Requested` creates a `Canceled` timeout with no payload. + +`Canceled` does not increment `Version`, so `Rescheduled` and `Canceled` can share a version. Rank breaks the tie, and terminal states win. + +`SagaTimeoutChangeRejected` never changes state. It adds a history entry to the timeout named by `TimeoutId`. If the store has no such timeout, it creates one with status `Unknown`. Lists exclude `Unknown` timeouts. They exist so that the change history and operator feedback have a record to attach to. + +### Saga completion + +`SagaCompleted` removes everything ServiceControl knows about the saga's timeouts: current state, versions, history, and failed triggers. It then writes a completed-saga marker keyed by `SagaId`. + +Any report for that saga that arrives after the marker is dropped. This covers a request reported in an earlier message that the primary processes after the completion. The marker expires with the retention period, so a report delayed for longer than that would create a stale `Pending` timeout. An operator can remove such a timeout with Forget. + +### History entries and deduplication + +Every message produces at most one history entry per timeout. Its deduplication key is: + +- `{Kind}:{Version}` for `Requested`, `Rescheduled`, `Canceled`, and `Triggered` +- `Rejected:{RequestedChange}:{ChangedAt ticks}` for `SagaTimeoutChangeRejected` + +Both persisters enforce the key, so applying the same message twice records one entry. + +### Failed triggers + +A timeout message that fails processing in the endpoint goes to the error queue. Error ingestion already stores it as a failed message with all headers. When a failed message carries `NServiceBus.SagaTimeoutId`, the ingestion unit of work also records a failed trigger: `{TimeoutId, TriggerId, FailedMessageId, FailedAt}`. This is one extra write per failed timeout message, and none for other messages. It is not a `MessageFailed` domain event handler, because that event does not carry the headers. + +A timeout has a failed trigger when its current `TriggerId` matches a recorded failed trigger whose failed message is still unresolved. A successful retry produces `SagaTimeoutTriggered`, and the timeout leaves `Pending`. An archived failed message leaves the timeout `Pending` with a failed trigger. The operator then decides whether to reschedule, trigger, or cancel it. + +### Overdue + +Overdue is computed when a query runs. It is not stored. A timeout is overdue when it is `Pending`, has no failed trigger, and `DueTime < now - SagaTimeouts/OverdueGracePeriod`. The default grace period is 5 minutes, which absorbs delayed-delivery latency and clock drift. An overdue timeout usually means one of the following: + +- the endpoint is down or its queue is backed up +- the delayed message was lost +- `SagaTimeoutTriggered` was lost, which can happen without the outbox + +## Persistence + +Both persisters implement `ISagaTimeoutDataStore`. The shared tests in `ServiceControl.Persistence.Tests` cover the apply rules above, including duplicates, reorder, and reports after completion. Each method owns its session or `DbContext`, commits before returning, and returns detached snapshots. This follows the coding guidelines. + +```csharp +interface ISagaTimeoutDataStore +{ + Task Apply(SagaTimeoutChange change, CancellationToken cancellationToken); + Task ApplySagaCompleted(SagaCompletedChange change, CancellationToken cancellationToken); + + Task>> Query(SagaTimeoutQuery query, PagingInfo paging, SortInfo sort, CancellationToken cancellationToken); + Task GetCounts(SagaTimeoutQuery query, CancellationToken cancellationToken); + Task> GetGroups(SagaTimeoutQuery query, SagaTimeoutGrouping grouping, CancellationToken cancellationToken); + Task Get(string timeoutId, CancellationToken cancellationToken); + Task> GetBySaga(Guid sagaId, CancellationToken cancellationToken); + Task>> QueryHistory(SagaTimeoutHistoryQuery query, PagingInfo paging, CancellationToken cancellationToken); + + Task RecordOperatorRequest(SagaTimeoutOperatorRequest request, CancellationToken cancellationToken); + Task Forget(string timeoutId, CancellationToken cancellationToken); +} +``` + +`Apply` returns `Ignored` for a report that follows the saga's completion marker. + +`SagaTimeoutQuery` carries: + +- the tab: `Pending`, `Overdue`, or `FailedTrigger` +- the endpoint, saga type, and saga id +- the due-time window +- the search text + +`SortInfo.AllowedSortOptions` for timeouts is `due_time`, `last_changed_at`, `endpoint`, `saga_type`, and `timeout_type`. + +`RecordOperatorRequest` checks the expected version and stores the pending operation on the timeout. It returns `Accepted`, `VersionMismatch`, `NotPending`, or `NotFound`. These are explicit outcomes, not exceptions, as the guidelines require for expected concurrency conflicts. + +### Logical model + +| Entity | Key | Holds | +|---|---|---| +| Saga timeout | `TimeoutId` | Current state: endpoint, saga type and id, correlation value, timeout type, version, trigger id, due time, status, requested and last-changed times, last change kind and source, management address, host, pending operator operation | +| Timeout version | `TimeoutId`, `Version` | Payload content type and payload of that version | +| History entry | `TimeoutId`, dedup key | Kind, version, trigger id, due time, `ChangedAt`, `ReceivedAt`, source, operation id, rejection details | +| Completed saga | `SagaId` | Endpoint, saga type, `ChangedAt`, `ReceivedAt` | +| Failed trigger | `TimeoutId`, `TriggerId` | Failed message id, failed at | + +The source of a change is `Operator` when the report carries an `OperationId`, and `Saga` otherwise. The management address is taken from the latest report. Payloads are stored per version, so the detail page can show "show change" between two versions. Lists never read payloads. + +The detail page shows "Payload updated" when a `Rescheduled` entry keeps the due time of the previous version. That label is computed when the history is read; there is no separate change kind. + +### RavenDB + +One document per timeout holds the current state and its history. The `Changes` list is bounded at 100 entries. When the bound is reached, the oldest entries except `Requested` are dropped and `HistoryTruncated` is set. Payloads are stored as document attachments named `payload-v{Version}`, so the document stays small and list queries do not load payloads. + +| Collection | Document id | Notes | +|---|---|---| +| `SagaTimeouts` | `SagaTimeouts/{TimeoutId}` | Current state, `Changes`, `PendingOperation`, `FailedTriggers` | +| `CompletedSagas` | `CompletedSagas/{SagaId}` | Completed-saga marker | + +Failed triggers are embedded in the timeout document and are rarely more than a few. Error ingestion patches the document with a `PatchRequest` that creates it as `Unknown` when it does not exist yet. That case happens when the failure is ingested before the `Requested` report. + +`Apply` loads the completed-saga marker and the timeout document in one session. If the marker exists, it returns `Ignored`. Otherwise it applies the rules and saves with `UseOptimisticConcurrency = true`. A `ConcurrencyException` is retried inside the method up to three times. Two concurrent reports for the same timeout are rare, so the retry almost never runs. + +`ApplySagaCompleted` stores the marker and deletes the saga's timeout documents in one session. Deleting a document also deletes its attachments. The documents to delete come from a query on `SagaTimeoutsIndex` by `SagaId`, which waits for non-stale results. A saga has few timeouts, so the session stays small. + +Indexes: + +- `SagaTimeoutsIndex` maps `Status`, `DueTime`, `EndpointName`, `SagaType`, `SagaId`, `TimeoutType`, `CorrelationValue`, `LastChangedAt`, and `HasUnresolvedFailedTrigger`. It also maps a `Query` field for full-text search over correlation value, saga type, and timeout type. The `Overdue` and `Pending` tabs are range queries on `DueTime` against a cutoff the API computes. +- `SagaTimeoutHistoryIndex` is a fan-out index: `from t in docs.SagaTimeouts from c in t.Changes select ...`. It stores the fields the Change History tab shows, so that tab reads the index without loading documents. + +`HasUnresolvedFailedTrigger` needs the failed message status, which lives in another collection. The index uses `LoadDocument` on `FailedMessages/{FailedMessageId}` for the current trigger only. RavenDB reindexes the timeout when that failed message changes. + +Retention uses `@expires`: + +- A timeout gets `@expires = terminal time + ErrorRetentionPeriod` when it becomes `Triggered` or `Canceled`. +- An `Unknown` timeout gets `@expires = ReceivedAt + ErrorRetentionPeriod`. A later report that makes it `Pending` removes the expiry with `ExpirationManager.CancelExpiration`. +- A completed-saga marker gets `@expires = ReceivedAt + ErrorRetentionPeriod`. +- `Pending` timeouts do not expire. + +A changed retention setting applies to new expiries only, as for other RavenDB data. + +### SQL Server and PostgreSQL + +The entities live in `ServiceControl.Persistence.EFCore` with explicit `IEntityTypeConfiguration` classes. Each provider gets one migration, `AddSagaTimeouts`, generated with `dotnet ef migrations add`. PostgreSQL names come from the snake-case convention. The names below are the PostgreSQL ones; SQL Server uses the PascalCase equivalents. + +```text +saga_timeouts + timeout_id varchar(450) PK + endpoint_name varchar(450) + saga_type varchar(450) + saga_id uuid + correlation_value varchar(450) null -- FitToIndex for long values + timeout_type varchar(450) + version int + trigger_id varchar(450) null + due_time timestamptz null + status smallint -- Unknown, Pending, Triggered, Canceled + rank smallint + requested_at timestamptz null + last_change_kind smallint + last_change_source smallint -- Saga, Operator + last_changed_at timestamptz + last_received_at timestamptz + status_changed_at timestamptz -- retention sweep key + management_address varchar(450) null + host varchar(450) null + host_id varchar(450) null + pending_operation_id uuid null + pending_operation jsonb / nvarchar(max) null + history_truncated bool + +saga_timeout_versions + timeout_id, version PK, FK -> saga_timeouts on delete cascade + payload_content_type varchar(450) + payload bytea / varbinary(max) + +saga_timeout_changes + id bigint identity PK + timeout_id varchar(450) FK -> saga_timeouts on delete cascade + dedup_key varchar(450) + kind, version, trigger_id, due_time, changed_at, received_at, source, operation_id, + requested_change, requested_due_time, reason + unique (timeout_id, dedup_key) + +saga_timeout_failed_triggers + timeout_id, trigger_id PK + failed_message_id varchar(450) + failed_at timestamptz + +completed_sagas + saga_id uuid PK + endpoint_name, saga_type, changed_at, received_at +``` + +`saga_timeout_failed_triggers` has no foreign key, because error ingestion can record a failure before the first report arrives. + +Indexes on `saga_timeouts`: + +- `(status, due_time)` for the Pending and Overdue tabs +- `(endpoint_name, status, due_time)` and `(saga_type, status, due_time)` for the filters and grouping +- `(saga_id)` for completion and the saga diagram +- `(correlation_value)` for search +- `(status, status_changed_at)` for the retention sweep + +`saga_timeout_changes` has an index on `(changed_at)` for the Change History tab. + +`Apply` runs in one transaction: + +1. Return `Ignored` if `completed_sagas` has the saga. +2. Insert the timeout row if it is absent, with status `Unknown`, version 0, and the saga fields from the change. +3. Insert the history entry and, for `Requested` and `Rescheduled`, the version row, each only if it is absent. +4. Update the current state with a guarded statement: `UPDATE saga_timeouts SET ... WHERE timeout_id = @id AND (version < @version OR (version = @version AND rank < @rank))`. + +Every insert uses the provider's insert-if-absent statement: `ON CONFLICT DO NOTHING` on PostgreSQL, and `INSERT ... WHERE NOT EXISTS` with `UPDLOCK, HOLDLOCK` on SQL Server. Two concurrent first reports therefore both succeed, and the guarded update decides which state wins. The guarded update replaces optimistic concurrency, because the version rule is the concurrency rule. The statements follow the `EndpointThroughput` upsert precedent (#5895), where duplicate keys are routine. A rejection stops after step 3, so a timeout known only from a rejection stays `Unknown`. + +Step 1 and `ApplySagaCompleted` can race. A report that passes step 1 just before the marker commits would leave a timeout behind. To prevent that, both take an application lock on the saga id for the transaction: `pg_advisory_xact_lock` on PostgreSQL and `sp_getapplock` on SQL Server. + +`ApplySagaCompleted` runs in one transaction. It inserts the marker if it is absent and deletes `saga_timeouts` rows for the saga. Versions and history go by cascade. It also deletes failed triggers for the deleted timeout ids. + +Search uses `LIKE` on `correlation_value`, `saga_type`, and `timeout_type`. Payload search is not in the first version. See [Open questions](#open-questions). + +`HasUnresolvedFailedTrigger` is a join from `saga_timeouts` to `saga_timeout_failed_triggers` on `(timeout_id, trigger_id)`, and on to `failed_messages` filtered to the unresolved status. Error ingestion inserts the failed trigger with the same insert-if-absent statement, through a new method on the ingestion unit of work and a statement in `IFailedMessageIngestionSqlDialect`. + +`RetentionSweeper` gets three passes: + +- Delete `saga_timeouts` whose status is `Triggered`, `Canceled`, or `Unknown` and whose `status_changed_at < now - ErrorRetentionPeriod`. Child rows go by cascade. +- Delete `completed_sagas` older than the same cutoff. +- Delete `saga_timeout_failed_triggers` with no matching timeout and older than the cutoff. + +The cutoff is recomputed on every run, so a changed setting applies to existing rows, as for failed messages. + +### Retention period + +Finished timeouts use `ErrorRetentionPeriod`. A timeout is error-instance operational data and lives about as long as the failed messages it can relate to. A separate `SagaTimeoutRetentionPeriod` setting is not proposed, but it can be added without a schema change. + +## HTTP API + +All routes are on the primary instance under `api/`. Paging, sorting, `Total-Count`, `Link`, and `ETag` work as in other list APIs. + +| Method and route | Permission | Purpose | +|---|---|---| +| `GET timeouts?tab=pending\|overdue\|failed&endpoint=&saga_type=&saga_id=&due_before=&due_after=&q=` | `error:timeouts:view` | Timeout list | +| `GET timeouts/counts` | `error:timeouts:view` | Counts per tab, for the tab headers and the menu badge | +| `GET timeouts/groups?by=endpoint\|saga_type&tab=...` | `error:timeouts:view` | Group rows with counts and earliest due time | +| `GET timeouts/history?endpoint=&saga_type=&since=` | `error:timeouts:view` | Change History tab | +| `GET timeouts/{timeoutId}` | `error:timeouts:view` | Details, history, and pending operation | +| `GET timeouts/{timeoutId}/payload?version=` | `error:timeouts:view` | Raw payload with its content type | +| `GET sagas/{sagaId}/timeouts` | `error:timeouts:view` | Timeouts of one saga instance, for the saga diagram | +| `POST timeouts/{timeoutId}/reschedule` | `error:timeouts:manage` | Body `{ "due_time": ..., "expected_version": n, "payload": ..., "payload_content_type": ... }`; the payload fields are optional | +| `POST timeouts/{timeoutId}/trigger` | `error:timeouts:manage` | Body `{ "expected_version": n }` | +| `POST timeouts/{timeoutId}/cancel` | `error:timeouts:manage` | Body `{ "expected_version": n }` | +| `DELETE timeouts/{timeoutId}` | `error:timeouts:manage` | Forget the record in ServiceControl only | + +The `GET sagas/{sagaId}/timeouts` route lives on the primary. `GET sagas/{sagaId}` is already a scatter-gather route on the primary. The new route reads only local data, so it does not need scatter-gather. + +The action routes return: + +- `202 Accepted` with `{ "operation_id": ... }` when the request is recorded and sent +- `409 Conflict` when `expected_version` does not match the stored version, or another operation is pending +- `422` when the timeout is not `Pending`, or has no known management address + +The check in ServiceControl gives fast feedback. It is not authoritative: the satellite repeats the version check against the saga data. + +`DELETE` exists for timeouts that will never be reported again, for example because the endpoint was decommissioned. It removes the record and its history from ServiceControl. It does not touch the saga. + +Every action writes an `IMessageActionAuditLog` entry, raises a domain event, and so appears in the event log. The events are `SagaTimeoutRescheduleRequested`, `SagaTimeoutTriggerRequested`, `SagaTimeoutCancelRequested`, `SagaTimeoutForgotten`, and `SagaTimeoutOperatorRequestRejected`. They relate to `/endpoint/{name}`. Integration events are not proposed for the first version. + +## Operator actions + +1. ServicePulse calls `POST timeouts/{id}/reschedule` with the version it displays. +2. The controller calls `RecordOperatorRequest`. On `Accepted`, it sends a local `RequestSagaTimeoutChange` command with `IMessageSession.SendLocal`, as `CustomCheckController` does for deletes, and returns `202`. +3. The handler for `RequestSagaTimeoutChange` builds the satellite request: the headers listed under [Management satellite](#management-satellite) and, for a payload change, the new payload as the body. It dispatches the request with `IMessageDispatcher` to the stored management address, inside the handler's transport transaction. `EditHandler` dispatches the same way. +4. The satellite applies the change and reports it with the `OperationId`. +5. The report clears the pending operation and records the history entry with source `Operator`. A rejection with the same `OperationId` clears it as well, records the reason, and raises `SagaTimeoutOperatorRequestRejected`. + +The trigger action sends a reschedule with the current time as the due time and an empty body. + +A pending operation that receives no report within `SagaTimeouts/OperatorRequestTimeout` (default 10 minutes) is shown as "no response from endpoint". It no longer blocks new requests. The record does not prove the request was lost: the endpoint can be down, or the request can sit in the satellite queue. + +The satellite queue is shared by all instances of a scaled-out endpoint, because its name derives from the endpoint's shared queue name. Any instance can apply the change, because the saga is in shared storage. + +## ServicePulse + +The new area follows existing patterns. It is gated on a minimum ServiceControl version with an `isSagaTimeoutsSupported.ts` file and `ConditionalRender`. `apiRoutes.ts` gets `viewTimeouts` and `manageTimeouts` capabilities. `PermissionGate` disables the action buttons for users without `error:timeouts:manage`. + +### Timeouts list + +The route is `/timeouts`, with child routes `pending`, `overdue`, `failed`, and `history`. A `TimeoutsMenuItem.vue` in `PageHeader.vue` shows the count of overdue timeouts plus failed triggers as a badge, read from `GET timeouts/counts`. There is no "NEW" badge component today. If product wants one, it is a small addition to the menu item styles. + +- The tabs use the `FailedMessagesView` pattern, with counts from `GET timeouts/counts`. +- The filter bar reuses `FilterInput` (debounced search), `ListFilterSelector` (endpoint, saga type), and a due-window selector based on `SuperDatePicker` and `timeRange.ts`, which already handle relative ranges. +- The table follows the mockup columns: Due, Timeout, Saga, Correlation value, Last change, Status. Due shows the absolute UTC time and a relative time. `TimeSince` already renders future times ("in 12 minutes"). `AdaptiveTimestamp` is past-only and needs a future-aware variant. +- Group by endpoint or saga type calls `GET timeouts/groups`. Selecting a group sets the filter. Grouping over the current page only would mislead, because it hides most of the data. +- Paging uses `PaginationStrip` and sorting uses `ColumnHeader`. +- A `TimeoutsStore` (Pinia) polls through `useStoreAutoRefresh`. After an action it switches to a 1-second interval until the pending operation clears, as `FailedMessages.vue` does after a retry. + +The Change History tab lists `GET timeouts/history` entries: + +- kind, timeout, saga, and version +- source: saga or operator +- the endpoint's `ChangedAt`, and ServiceControl's `ReceivedAt` when they differ by more than the grace period + +### Timeout details + +The route is `/timeouts/{timeoutId}`, with breadcrumbs Timeouts / endpoint / saga type / correlation value. + +- The details panel shows the stored fields, the host, and the current trigger id. The version shown is the integer `Version`, not a fragment of the trigger id as in the mockup. It is the value sent as `expected_version`. +- The payload panel renders JSON payloads with the existing message body viewer. Other content types show as text or as a download. +- The history panel is a timeline of history entries. "Show change" on a `Rescheduled` entry shows a due-time change and a payload diff against the previous version. +- The actions panel has Reschedule, Trigger now, and Cancel. Reschedule has a UTC date-time input and an optional payload editor for JSON payloads. Cancel uses `ConfirmDialog`. All three send the displayed version. +- A `409` shows a toast saying the timeout changed and reloads the page. +- While an operation is pending, the panel shows "Change requested, waiting for the endpoint", with the time of the request, and disables the actions. +- The related panel lists the other timeouts of the saga instance and links to the saga diagram. The diagram link needs a message id, which the page takes from the saga history (`GET sagas/{sagaId}`). Without saga audit, the link is not shown. +- Forget is a secondary action, shown only for `Unknown` timeouts and timeouts whose operation got no response. + +### Saga diagram additions + +`SagaDiagram.vue` adds a "Timeouts (n)" toggle to its toolbar. `SagaDiagramStore` loads `GET sagas/{sagaId}/timeouts` next to `GET sagas/{sagaId}` when the feature is supported. A failure of the new call does not break the diagram. A completed saga has no timeouts left in ServiceControl, so the panel is empty for it. + +- The "Timeouts of this saga" panel lists each timeout with status, requested, rescheduled, and triggered times, and links to Open in Timeouts and Show history. +- `SagaDiagramParser` treats an outgoing message with a delivery delay as a timeout request. That message is matched to a managed timeout by saga id, timeout type, and due time (`deliver_at` within one second of `DueTime`). The match adds a "Managed" badge and a "View timeout" link. +- The match is a heuristic, because saga audit does not record the `NServiceBus.SagaTimeoutId` header. Two timeouts of the same type with the same due time cannot be told apart. Adding the header to the saga audit `ResultingMessage` would make the match exact; that is a change to NServiceBus.SagaAudit and the audit instance. +- Operator changes are not saga invocations, so saga audit never records them. The diagram inserts them from the timeout history as separate timeline entries, ordered by `ChangedAt` ("Timeout Rescheduled = 1m, by an operator in ServicePulse"). + +### Testing + +- Component tests use the `SagaDiagram.spec.ts` style and a timeout history fixture. +- Application tests use MSW handlers through `mockEndpoint` and a `hasSagaTimeouts` precondition. The precondition sets the minimum ServiceControl version. +- A `timeouts` mock scenario in `test/mocks/scenarios` supports manual work with `npm run dev:mocks`. + +## Rollout and compatibility + +Reporting is opt-in because a ServiceControl version without this feature does not know the six message types. Reports sent to it would fail on the primary's input queue and end up in the error queue as failed messages of ServiceControl itself. The documentation of `ReportSagaTimeoutsToServiceControl()` states the minimum ServiceControl version. + +Enabling reporting adds the `{endpoint queue}.SagaTimeouts` queue. Installers create it like the endpoint's other queues. Deployments that create queues with scripts must add it. + +Endpoints with persisters that do not implement `ISupportTimeoutManagement` keep the legacy timeout behavior and report nothing. Their timeouts do not appear in ServicePulse. The satellite still runs on such an endpoint and rejects every request with `TimeoutNotFound`. + +Every timeout change is one message to the primary instance. A saga that requests a timeout per message doubles the primary's plugin-message load for that endpoint. The load test tooling in this repository should get a saga timeout scenario before release. It should measure handler throughput on both persisters with concurrent reports for the same saga. + +The feature is not available on the audit or monitoring instances, and it is not available when the primary runs only error ingestion. + +## Open questions + +| Question | Owner | +|---|---| +| The minimum ServiceControl version to state in the `ReportSagaTimeoutsToServiceControl()` documentation | NServiceBus, ServiceControl | +| Record `NServiceBus.SagaTimeoutId` in saga audit outgoing messages, so the saga diagram match is exact | NServiceBus.SagaAudit, ServiceControl audit | +| Payload search: a full-text column on the version table, or the RavenDB `Query` field over JSON payloads, and its cost | ServiceControl | +| Whether finished timeouts need their own retention setting | ServiceControl | +| Whether the losses possible without the outbox, a lost `Triggered` report and a lost dispatch after a satellite commit, need endpoint-side detection, or whether showing them as overdue is enough | NServiceBus, ServiceControl | diff --git a/src/ServiceControl.AcceptanceTests.RavenDB/SagaTimeouts/When_a_saga_timeout_is_requested.cs b/src/ServiceControl.AcceptanceTests.RavenDB/SagaTimeouts/When_a_saga_timeout_is_requested.cs new file mode 100644 index 0000000000..663e283790 --- /dev/null +++ b/src/ServiceControl.AcceptanceTests.RavenDB/SagaTimeouts/When_a_saga_timeout_is_requested.cs @@ -0,0 +1,89 @@ +namespace ServiceControl.AcceptanceTests.RavenDB.SagaTimeouts +{ + using System; + using System.Collections.Generic; + using System.Linq; + using System.Text; + using System.Threading.Tasks; + using AcceptanceTesting; + using AcceptanceTesting.EndpointTemplates; + using NServiceBus; + using NServiceBus.AcceptanceTesting; + using NServiceBus.Routing; + using NServiceBus.Transport; + using NUnit.Framework; + using Persistence.SagaTimeouts; + using ServiceBus.Management.Infrastructure.Settings; + + class When_a_saga_timeout_is_requested : AcceptanceTest + { + [Test] + public async Task Should_list_it_as_pending() + { + SagaTimeoutView listed = null; + SagaTimeoutDetails details = null; + + await Define() + .WithEndpoint() + .Do("Wait for the timeout to be listed", async ctx => + { + var timeouts = await this.TryGetMany("/api/timeouts?status=pending", timeout => timeout.TimeoutId == ctx.TimeoutId); + listed = timeouts.HasResult ? timeouts.Items.Single() : null; + return listed != null; + }) + .Do("Read its details", async ctx => + { + var result = await this.TryGet($"/api/timeouts/{ctx.TimeoutId}"); + details = result.HasResult ? result.Item : null; + return details != null; + }) + .Done(_ => true) + .Run(); + + Assert.Multiple(() => + { + Assert.That(listed.Status, Is.EqualTo(SagaTimeoutStatus.Pending)); + Assert.That(listed.Version, Is.EqualTo(1)); + Assert.That(listed.ManagementAddress, Is.EqualTo("Sales.SagaTimeouts")); + Assert.That(details.Payload, Is.EqualTo("{\"OrderId\":\"order-1\"}")); + Assert.That(details.History.Single().Kind, Is.EqualTo(SagaTimeoutChangeKind.Requested)); + }); + } + + class Context : ScenarioContext, ISequenceContext + { + public int Step { get; set; } + public string TimeoutId { get; } = Guid.NewGuid().ToString(); + } + + public class Sales : EndpointConfigurationBuilder + { + public Sales() => EndpointSetup(c => c.EnableFeature()); + + // Sent as NServiceBus sends it: the enclosed type is the full name without an assembly, and the body is System.Text.Json + class ReportTimeoutRequest : DispatchRawMessages + { + protected override TransportOperations CreateMessage(Context context) + { + var payload = Convert.ToBase64String(Encoding.UTF8.GetBytes("{\"OrderId\":\"order-1\"}")); + var body = $$""" + {"TriggerId":"trigger-1","DueTime":"{{DateTimeOffset.UtcNow.AddHours(1):O}}","PayloadContentType":"application/json","Payload":"{{payload}}", + "TimeoutId":"{{context.TimeoutId}}","TimeoutType":"Sales.PaymentTimeout","Version":1,"EndpointName":"Sales","SagaType":"Sales.OrderSaga", + "SagaId":"{{Guid.NewGuid()}}","CorrelationPropertyValue":"order-1","ChangedAt":"{{DateTimeOffset.UtcNow:O}}", + "Host":"host","HostId":"host-id","ManagementAddress":"Sales.SagaTimeouts"} + """; + + var headers = new Dictionary + { + [Headers.EnclosedMessageTypes] = "NServiceBus.SagaTimeoutRequested", + [Headers.ContentType] = "application/json", + [Headers.MessageIntent] = "Send" + }; + + var message = new OutgoingMessage(Guid.NewGuid().ToString(), headers, Encoding.UTF8.GetBytes(body)); + return new TransportOperations(new TransportOperation(message, new UnicastAddressTag(Settings.DEFAULT_INSTANCE_NAME))); + } + } + } + } +} diff --git a/src/ServiceControl.Infrastructure/Auth/MessageAction.cs b/src/ServiceControl.Infrastructure/Auth/MessageAction.cs index f8463e5eac..fb6842336e 100644 --- a/src/ServiceControl.Infrastructure/Auth/MessageAction.cs +++ b/src/ServiceControl.Infrastructure/Auth/MessageAction.cs @@ -8,7 +8,10 @@ public enum MessageActionKind Archive, Unarchive, Edit, - Delete + Delete, + Reschedule, + Trigger, + Cancel } /// How the action selected the messages it acts on. diff --git a/src/ServiceControl.Infrastructure/Auth/Permissions.cs b/src/ServiceControl.Infrastructure/Auth/Permissions.cs index 45e9cec07f..b9c9375e6e 100644 --- a/src/ServiceControl.Infrastructure/Auth/Permissions.cs +++ b/src/ServiceControl.Infrastructure/Auth/Permissions.cs @@ -55,6 +55,11 @@ public static class Permissions /// Sagas area — viewing saga audit data. public const string ErrorSagasView = "error:sagas:view"; + /// Saga timeouts area — viewing saga timeouts and requesting changes to them. + public const string ErrorTimeoutsView = "error:timeouts:view"; + /// + public const string ErrorTimeoutsManage = "error:timeouts:manage"; + /// Event log area — viewing the event log. public const string ErrorEventLogView = "error:eventlog:view"; diff --git a/src/ServiceControl.Infrastructure/Auth/RolePermissions.cs b/src/ServiceControl.Infrastructure/Auth/RolePermissions.cs index b1c273ee27..a7469d81ea 100644 --- a/src/ServiceControl.Infrastructure/Auth/RolePermissions.cs +++ b/src/ServiceControl.Infrastructure/Auth/RolePermissions.cs @@ -19,6 +19,7 @@ public static class RolePermissions Permissions.ErrorHeartbeatsView, Permissions.ErrorCustomChecksView, Permissions.ErrorSagasView, + Permissions.ErrorTimeoutsView, Permissions.ErrorEventLogView, Permissions.ErrorQueuesView, Permissions.ErrorConnectionsView, @@ -50,6 +51,7 @@ public static class RolePermissions Permissions.ErrorEndpointsManage, Permissions.ErrorEndpointsDelete, Permissions.ErrorCustomChecksDelete, + Permissions.ErrorTimeoutsManage, Permissions.ErrorQueuesDelete, Permissions.ErrorConnectionsManage, Permissions.MonitoringEndpointDelete, diff --git a/src/ServiceControl.Persistence.EFCore/Abstractions/BasePersistence.cs b/src/ServiceControl.Persistence.EFCore/Abstractions/BasePersistence.cs index 4b76e5add7..47983799fd 100644 --- a/src/ServiceControl.Persistence.EFCore/Abstractions/BasePersistence.cs +++ b/src/ServiceControl.Persistence.EFCore/Abstractions/BasePersistence.cs @@ -15,6 +15,7 @@ namespace ServiceControl.Persistence.EFCore.Abstractions; using ServiceControl.Persistence.EFCore.Infrastructure.Metrics; using ServiceControl.Persistence.MessageRedirects; using ServiceControl.Persistence.Recoverability; +using ServiceControl.Persistence.SagaTimeouts; using ServiceControl.Persistence.UnitOfWork; using ServiceControl.Recoverability; using ServiceControl.Recoverability.Archiving.Metrics; @@ -73,6 +74,7 @@ protected static void RegisterDataStores(IServiceCollection services, EFPersiste services.AddSingleton(); services.AddSingleton(); services.AddSingleton(); + services.AddSingleton(); services.AddSingleton(); diff --git a/src/ServiceControl.Persistence.EFCore/Implementation/SagaTimeoutDataStore.cs b/src/ServiceControl.Persistence.EFCore/Implementation/SagaTimeoutDataStore.cs new file mode 100644 index 0000000000..1a4bbeed23 --- /dev/null +++ b/src/ServiceControl.Persistence.EFCore/Implementation/SagaTimeoutDataStore.cs @@ -0,0 +1,26 @@ +namespace ServiceControl.Persistence.EFCore.Implementation; + +using ServiceControl.Persistence.Infrastructure; +using ServiceControl.Persistence.SagaTimeouts; + +// Saga timeout management is implemented for RavenDB only so far +public class SagaTimeoutDataStore : ISagaTimeoutDataStore +{ + public Task Apply(SagaTimeoutChange change, CancellationToken cancellationToken = default) => throw NotSupported(); + + public Task ApplySagaCompleted(SagaCompletedChange change, CancellationToken cancellationToken = default) => throw NotSupported(); + + public Task>> Query(SagaTimeoutQuery query, PagingInfo paging, SortInfo sort, CancellationToken cancellationToken = default) => throw NotSupported(); + + public Task GetCounts(CancellationToken cancellationToken = default) => throw NotSupported(); + + public Task Get(string timeoutId, CancellationToken cancellationToken = default) => throw NotSupported(); + + public Task> GetBySaga(Guid sagaId, CancellationToken cancellationToken = default) => throw NotSupported(); + + public Task RecordOperatorRequest(SagaTimeoutOperatorRequest request, CancellationToken cancellationToken = default) => throw NotSupported(); + + public Task Forget(string timeoutId, CancellationToken cancellationToken = default) => throw NotSupported(); + + static NotSupportedException NotSupported() => new("Saga timeout management is not supported by this persister yet."); +} diff --git a/src/ServiceControl.Persistence.RavenDB/ExpirationManager.cs b/src/ServiceControl.Persistence.RavenDB/ExpirationManager.cs index 7a8040de71..005f39b234 100644 --- a/src/ServiceControl.Persistence.RavenDB/ExpirationManager.cs +++ b/src/ServiceControl.Persistence.RavenDB/ExpirationManager.cs @@ -34,6 +34,19 @@ public void EnableExpiration(IAsyncDocumentSession session, FailedMessage failed session.Advanced.GetMetadataFor(failedMessage)[Constants.Documents.Metadata.Expires] = expiresAt; } + // For error-instance data other than failed messages, such as finished saga timeouts + public void EnableErrorRetentionExpiration(IAsyncDocumentSession session, object document) + { + var expiresAt = timeProvider.GetUtcNow().UtcDateTime + errorRetentionPeriod; + + session.Advanced.GetMetadataFor(document)[Constants.Documents.Metadata.Expires] = expiresAt; + } + + public void CancelExpiration(IAsyncDocumentSession session, object document) + { + session.Advanced.GetMetadataFor(document).Remove(Constants.Documents.Metadata.Expires); + } + public void EnableExpiration(IAsyncDocumentSession session, EventLogItem eventLogItem) { var expiresAt = timeProvider.GetUtcNow().UtcDateTime + eventsRetentionPeriod; diff --git a/src/ServiceControl.Persistence.RavenDB/RavenPersistence.cs b/src/ServiceControl.Persistence.RavenDB/RavenPersistence.cs index 95dc171a86..9bcad385f7 100644 --- a/src/ServiceControl.Persistence.RavenDB/RavenPersistence.cs +++ b/src/ServiceControl.Persistence.RavenDB/RavenPersistence.cs @@ -13,8 +13,10 @@ namespace ServiceControl.Persistence.RavenDB; using Persistence.MessageRedirects; using Persistence.Recoverability; using Recoverability; +using SagaTimeouts; using ServiceControl.CustomChecks; using ServiceControl.Infrastructure.RavenDB.Subscriptions; +using ServiceControl.Persistence.SagaTimeouts; using ServiceControl.Recoverability; using Throughput; using UnitOfWork; @@ -75,6 +77,7 @@ public void AddPersistence(IServiceCollection services) services.AddSingleton(); services.AddSingleton(); services.AddSingleton(); + services.AddSingleton(); } public void AddInstaller(IServiceCollection services) diff --git a/src/ServiceControl.Persistence.RavenDB/SagaTimeouts/SagaTimeout.cs b/src/ServiceControl.Persistence.RavenDB/SagaTimeouts/SagaTimeout.cs new file mode 100644 index 0000000000..d9cdc23807 --- /dev/null +++ b/src/ServiceControl.Persistence.RavenDB/SagaTimeouts/SagaTimeout.cs @@ -0,0 +1,87 @@ +namespace ServiceControl.Persistence.RavenDB.SagaTimeouts +{ + using System; + using System.Collections.Generic; + using ServiceControl.Persistence.SagaTimeouts; + + // The type names give the collection names SagaTimeouts and CompletedSagas + class SagaTimeout + { + public string Id { get; set; } + public string TimeoutId { get; set; } + public string EndpointName { get; set; } + public string SagaType { get; set; } + public Guid SagaId { get; set; } + public string CorrelationValue { get; set; } + public string TimeoutType { get; set; } + public int Version { get; set; } + public int Rank { get; set; } + public string TriggerId { get; set; } + public DateTime? DueTime { get; set; } + public SagaTimeoutStatus Status { get; set; } + public DateTime? RequestedAt { get; set; } + public SagaTimeoutChangeKind LastChangeKind { get; set; } + public SagaTimeoutChangeSource LastChangeSource { get; set; } + public DateTime? LastChangedAt { get; set; } + public DateTime? StatusChangedAt { get; set; } + + // ChangedAt of the newest report, which decides whether its host and management address replace the stored ones + public DateTime? LastReportedAt { get; set; } + public string Host { get; set; } + public string HostId { get; set; } + public string ManagementAddress { get; set; } + + public PendingOperation PendingOperation { get; set; } + public bool HistoryTruncated { get; set; } + public List Changes { get; set; } = []; + + // The payload of each version is an attachment named by PayloadAttachmentName + public List Payloads { get; set; } = []; + + public static string MakeDocumentId(string timeoutId) => $"SagaTimeouts/{timeoutId}"; + + public static string PayloadAttachmentName(int version) => $"payload-v{version}"; + } + + class PendingOperation + { + public Guid OperationId { get; set; } + public SagaTimeoutOperation Change { get; set; } + public DateTime RequestedAt { get; set; } + public DateTime? RequestedDueTime { get; set; } + } + + class HistoryEntry + { + public string DeduplicationKey { get; set; } + public SagaTimeoutChangeKind Kind { get; set; } + public int Version { get; set; } + public string TriggerId { get; set; } + public DateTime? DueTime { get; set; } + public DateTime ChangedAt { get; set; } + public DateTime ReceivedAt { get; set; } + public SagaTimeoutChangeSource Source { get; set; } + public Guid? OperationId { get; set; } + public string RequestedChange { get; set; } + public DateTime? RequestedDueTime { get; set; } + public string Reason { get; set; } + } + + class PayloadVersion + { + public int Version { get; set; } + public string ContentType { get; set; } + } + + class CompletedSaga + { + public string Id { get; set; } + public Guid SagaId { get; set; } + public string EndpointName { get; set; } + public string SagaType { get; set; } + public DateTime ChangedAt { get; set; } + public DateTime ReceivedAt { get; set; } + + public static string MakeDocumentId(Guid sagaId) => $"CompletedSagas/{sagaId}"; + } +} diff --git a/src/ServiceControl.Persistence.RavenDB/SagaTimeouts/SagaTimeoutDataStore.cs b/src/ServiceControl.Persistence.RavenDB/SagaTimeouts/SagaTimeoutDataStore.cs new file mode 100644 index 0000000000..f89eb13692 --- /dev/null +++ b/src/ServiceControl.Persistence.RavenDB/SagaTimeouts/SagaTimeoutDataStore.cs @@ -0,0 +1,485 @@ +namespace ServiceControl.Persistence.RavenDB.SagaTimeouts +{ + using System; + using System.Collections.Generic; + using System.IO; + using System.Linq; + using System.Linq.Expressions; + using System.Text; + using System.Threading; + using System.Threading.Tasks; + using Raven.Client.Documents; + using Raven.Client.Documents.Linq; + using Raven.Client.Documents.Queries; + using Raven.Client.Documents.Session; + using Raven.Client.Exceptions; + using ServiceControl.Persistence.Infrastructure; + using ServiceControl.Persistence.SagaTimeouts; + + class SagaTimeoutDataStore(IRavenSessionProvider sessionProvider, ExpirationManager expirationManager, TimeProvider timeProvider) : ISagaTimeoutDataStore + { + // Two concurrent reports for the same timeout are rare, so a conflict almost never repeats + const int MaxConcurrencyRetries = 3; + + static readonly TimeSpan CompletionQueryWait = TimeSpan.FromSeconds(30); + + public async Task Apply(SagaTimeoutChange change, CancellationToken cancellationToken = default) + { + for (var attempt = 0; ; attempt++) + { + try + { + return await TryApply(change, cancellationToken); + } + catch (ConcurrencyException) when (attempt < MaxConcurrencyRetries) + { + } + } + } + + async Task TryApply(SagaTimeoutChange change, CancellationToken cancellationToken) + { + using var session = await sessionProvider.OpenSession(cancellationToken: cancellationToken); + session.Advanced.UseOptimisticConcurrency = true; + + var marker = await session.LoadAsync(CompletedSaga.MakeDocumentId(change.SagaId), cancellationToken); + if (marker != null) + { + return SagaTimeoutApplyOutcome.Ignored; + } + + var documentId = SagaTimeout.MakeDocumentId(change.TimeoutId); + var timeout = await session.LoadAsync(documentId, cancellationToken); + var isNew = timeout == null; + + if (isNew) + { + timeout = new SagaTimeout + { + Id = documentId, + TimeoutId = change.TimeoutId, + Status = SagaTimeoutStatus.Unknown, + Rank = -1 + }; + await session.StoreAsync(timeout, cancellationToken); + } + + var deduplicationKey = SagaTimeoutRules.DeduplicationKey(change); + if (timeout.Changes.Any(entry => entry.DeduplicationKey == deduplicationKey)) + { + return SagaTimeoutApplyOutcome.Duplicate; + } + + timeout.EndpointName ??= change.EndpointName; + timeout.SagaType ??= change.SagaType; + timeout.SagaId = change.SagaId; + timeout.CorrelationValue ??= change.CorrelationValue; + timeout.TimeoutType ??= change.TimeoutType; + + var changedAt = change.ChangedAt.UtcDateTime; + if (timeout.LastReportedAt == null || changedAt >= timeout.LastReportedAt) + { + timeout.LastReportedAt = changedAt; + timeout.Host = change.Host ?? timeout.Host; + timeout.HostId = change.HostId ?? timeout.HostId; + timeout.ManagementAddress = change.ManagementAddress ?? timeout.ManagementAddress; + } + + AddHistoryEntry(timeout, change, deduplicationKey); + + if (change.Kind == SagaTimeoutChangeKind.Requested && timeout.RequestedAt == null) + { + timeout.RequestedAt = changedAt; + } + + if (change.Payload != null && timeout.Payloads.All(payload => payload.Version != change.Version)) + { + timeout.Payloads.Add(new PayloadVersion { Version = change.Version, ContentType = change.PayloadContentType }); + session.Advanced.Attachments.Store(timeout, SagaTimeout.PayloadAttachmentName(change.Version), new MemoryStream(change.Payload), change.PayloadContentType); + } + + if (change.OperationId.HasValue && timeout.PendingOperation?.OperationId == change.OperationId) + { + timeout.PendingOperation = null; + } + + var outcome = SagaTimeoutApplyOutcome.HistoryOnly; + if (change.Kind != SagaTimeoutChangeKind.Rejected && IsNewer(change, timeout)) + { + ApplyState(timeout, change); + outcome = SagaTimeoutApplyOutcome.Applied; + } + + if (timeout.Status == SagaTimeoutStatus.Pending) + { + expirationManager.CancelExpiration(session, timeout); + } + else if (isNew || outcome == SagaTimeoutApplyOutcome.Applied) + { + // Finished and Unknown timeouts live as long as the failed messages they can relate to + expirationManager.EnableErrorRetentionExpiration(session, timeout); + } + + await session.SaveChangesAsync(cancellationToken); + + return outcome; + } + + static bool IsNewer(SagaTimeoutChange change, SagaTimeout timeout) + { + var rank = SagaTimeoutRules.RankOf(change.Kind); + return change.Version > timeout.Version || (change.Version == timeout.Version && rank > timeout.Rank); + } + + static void ApplyState(SagaTimeout timeout, SagaTimeoutChange change) + { + timeout.Version = change.Version; + timeout.Rank = SagaTimeoutRules.RankOf(change.Kind); + timeout.Status = SagaTimeoutRules.StatusAfter(change.Kind); + timeout.TimeoutType = change.TimeoutType ?? timeout.TimeoutType; + timeout.TriggerId = change.TriggerId ?? timeout.TriggerId; + timeout.DueTime = change.DueTime?.UtcDateTime ?? timeout.DueTime; + timeout.LastChangeKind = change.Kind; + timeout.LastChangeSource = change.Source; + timeout.LastChangedAt = change.ChangedAt.UtcDateTime; + timeout.StatusChangedAt = change.ReceivedAt.UtcDateTime; + } + + static void AddHistoryEntry(SagaTimeout timeout, SagaTimeoutChange change, string deduplicationKey) + { + timeout.Changes.Add(new HistoryEntry + { + DeduplicationKey = deduplicationKey, + Kind = change.Kind, + Version = change.Version, + TriggerId = change.TriggerId, + DueTime = change.DueTime?.UtcDateTime, + ChangedAt = change.ChangedAt.UtcDateTime, + ReceivedAt = change.ReceivedAt.UtcDateTime, + Source = change.Source, + OperationId = change.OperationId, + RequestedChange = change.RequestedChange, + RequestedDueTime = change.RequestedDueTime?.UtcDateTime, + Reason = change.Reason + }); + + while (timeout.Changes.Count > SagaTimeoutRules.MaxHistoryEntries) + { + var oldest = timeout.Changes + .Where(entry => entry.Kind != SagaTimeoutChangeKind.Requested) + .OrderBy(entry => entry.ChangedAt) + .First(); + + timeout.Changes.Remove(oldest); + timeout.HistoryTruncated = true; + } + } + + public async Task ApplySagaCompleted(SagaCompletedChange change, CancellationToken cancellationToken = default) + { + using var session = await sessionProvider.OpenSession(cancellationToken: cancellationToken); + + var markerId = CompletedSaga.MakeDocumentId(change.SagaId); + var marker = await session.LoadAsync(markerId, cancellationToken); + var outcome = SagaTimeoutApplyOutcome.Duplicate; + + if (marker == null) + { + marker = new CompletedSaga + { + Id = markerId, + SagaId = change.SagaId, + EndpointName = change.EndpointName, + SagaType = change.SagaType, + ChangedAt = change.ChangedAt.UtcDateTime, + ReceivedAt = change.ReceivedAt.UtcDateTime + }; + await session.StoreAsync(marker, cancellationToken); + expirationManager.EnableErrorRetentionExpiration(session, marker); + outcome = SagaTimeoutApplyOutcome.Applied; + } + + // A redelivered completion still deletes timeouts that a report created after the first one + var timeouts = await session.Query() + .Customize(x => x.WaitForNonStaleResults(CompletionQueryWait)) + .Where(timeout => timeout.SagaId == change.SagaId) + .OfType() + .ToListAsync(cancellationToken); + + foreach (var timeout in timeouts) + { + session.Delete(timeout); + } + + await session.SaveChangesAsync(cancellationToken); + + return outcome; + } + + public async Task>> Query(SagaTimeoutQuery query, PagingInfo paging, SortInfo sort, CancellationToken cancellationToken = default) + { + var now = timeProvider.GetUtcNow().UtcDateTime; + + using var session = await sessionProvider.OpenSession(cancellationToken: cancellationToken); + + var ravenQuery = session.Query() + .Statistics(out var stats) + .FilterByStatus(query.Status, now - SagaTimeoutRules.OverdueGracePeriod); + + if (!string.IsNullOrWhiteSpace(query.EndpointName)) + { + ravenQuery = ravenQuery.Where(timeout => timeout.EndpointName == query.EndpointName); + } + + if (!string.IsNullOrWhiteSpace(query.SagaType)) + { + ravenQuery = ravenQuery.Where(timeout => timeout.SagaType == query.SagaType); + } + + if (query.SagaId.HasValue) + { + ravenQuery = ravenQuery.Where(timeout => timeout.SagaId == query.SagaId.Value); + } + + if (!string.IsNullOrWhiteSpace(query.SearchText)) + { + var terms = query.SearchText.Split(' ', StringSplitOptions.RemoveEmptyEntries | StringSplitOptions.TrimEntries) + .Select(term => $"*{term}*"); + ravenQuery = ravenQuery.Search(timeout => timeout.Query, string.Join(" ", terms), options: SearchOptions.And, @operator: SearchOperator.And); + } + + var results = await Sort(ravenQuery, sort) + .Skip(paging.Offset) + .Take(paging.PageSize) + .OfType() + .ToListAsync(cancellationToken); + + IList views = results.Select(timeout => ToView(timeout, now)).ToList(); + return new QueryResult>(views, stats.ToQueryStatsInfo()); + } + + static IRavenQueryable Sort(IRavenQueryable query, SortInfo sort) + { + Expression> keySelector = sort.Sort switch + { + "last_changed_at" => timeout => timeout.LastChangedAt, + "endpoint" => timeout => timeout.EndpointName, + "saga_type" => timeout => timeout.SagaType, + "timeout_type" => timeout => timeout.TimeoutType, + _ => timeout => timeout.DueTime + }; + + return sort.Direction == "desc" ? query.OrderByDescending(keySelector) : query.OrderBy(keySelector); + } + + public async Task GetCounts(CancellationToken cancellationToken = default) + { + var overdueCutoff = timeProvider.GetUtcNow().UtcDateTime - SagaTimeoutRules.OverdueGracePeriod; + + using var session = await sessionProvider.OpenSession(cancellationToken: cancellationToken); + + var pending = await session.Query() + .FilterByStatus(SagaTimeoutListStatus.Pending, overdueCutoff) + .CountAsync(cancellationToken); + var overdue = await session.Query() + .FilterByStatus(SagaTimeoutListStatus.Overdue, overdueCutoff) + .CountAsync(cancellationToken); + + return new SagaTimeoutCounts { Pending = pending, Overdue = overdue }; + } + + public async Task Get(string timeoutId, CancellationToken cancellationToken = default) + { + using var session = await sessionProvider.OpenSession(cancellationToken: cancellationToken); + + var documentId = SagaTimeout.MakeDocumentId(timeoutId); + var timeout = await session.LoadAsync(documentId, cancellationToken); + if (timeout == null) + { + return null; + } + + string payload = null; + string payloadContentType = null; + var currentPayload = timeout.Payloads.FirstOrDefault(p => p.Version == timeout.Version); + if (currentPayload != null) + { + using var attachment = await session.Advanced.Attachments.GetAsync(documentId, SagaTimeout.PayloadAttachmentName(currentPayload.Version), cancellationToken); + if (attachment != null) + { + using var reader = new StreamReader(attachment.Stream, Encoding.UTF8); + payload = await reader.ReadToEndAsync(cancellationToken); + payloadContentType = currentPayload.ContentType; + } + } + + var now = timeProvider.GetUtcNow().UtcDateTime; + return new SagaTimeoutDetails + { + TimeoutId = timeout.TimeoutId, + EndpointName = timeout.EndpointName, + SagaType = timeout.SagaType, + SagaId = timeout.SagaId, + CorrelationValue = timeout.CorrelationValue, + TimeoutType = timeout.TimeoutType, + Version = timeout.Version, + TriggerId = timeout.TriggerId, + DueTime = timeout.DueTime, + Status = timeout.Status, + IsOverdue = SagaTimeoutRules.IsOverdue(timeout.Status, timeout.DueTime, now), + RequestedAt = timeout.RequestedAt, + LastChangeKind = timeout.LastChangeKind, + LastChangeSource = timeout.LastChangeSource, + LastChangedAt = timeout.LastChangedAt, + Host = timeout.Host, + ManagementAddress = timeout.ManagementAddress, + PendingOperation = ToView(timeout.PendingOperation), + PayloadContentType = payloadContentType, + Payload = payload, + HistoryTruncated = timeout.HistoryTruncated, + History = timeout.Changes + .OrderBy(entry => entry.ChangedAt) + .Select(entry => new SagaTimeoutHistoryEntryView + { + Kind = entry.Kind, + Version = entry.Version, + TriggerId = entry.TriggerId, + DueTime = entry.DueTime, + ChangedAt = entry.ChangedAt, + ReceivedAt = entry.ReceivedAt, + Source = entry.Source, + OperationId = entry.OperationId, + RequestedChange = entry.RequestedChange, + RequestedDueTime = entry.RequestedDueTime, + Reason = entry.Reason + }) + .ToList() + }; + } + + public async Task> GetBySaga(Guid sagaId, CancellationToken cancellationToken = default) + { + var now = timeProvider.GetUtcNow().UtcDateTime; + + using var session = await sessionProvider.OpenSession(cancellationToken: cancellationToken); + + var timeouts = await session.Query() + .Where(timeout => timeout.SagaId == sagaId && timeout.Status != SagaTimeoutStatus.Unknown) + .OrderBy(timeout => timeout.DueTime) + .OfType() + .ToListAsync(cancellationToken); + + return timeouts.Select(timeout => ToView(timeout, now)).ToList(); + } + + public async Task RecordOperatorRequest(SagaTimeoutOperatorRequest request, CancellationToken cancellationToken = default) + { + for (var attempt = 0; ; attempt++) + { + try + { + return await TryRecordOperatorRequest(request, cancellationToken); + } + catch (ConcurrencyException) when (attempt < MaxConcurrencyRetries) + { + } + } + } + + async Task TryRecordOperatorRequest(SagaTimeoutOperatorRequest request, CancellationToken cancellationToken) + { + var now = timeProvider.GetUtcNow().UtcDateTime; + + using var session = await sessionProvider.OpenSession(cancellationToken: cancellationToken); + session.Advanced.UseOptimisticConcurrency = true; + + var timeout = await session.LoadAsync(SagaTimeout.MakeDocumentId(request.TimeoutId), cancellationToken); + if (timeout == null) + { + return new SagaTimeoutOperatorRequestResult(SagaTimeoutOperatorRequestOutcome.NotFound); + } + + var outcome = SagaTimeoutOperatorRequestOutcome.Accepted; + if (timeout.Status != SagaTimeoutStatus.Pending) + { + outcome = SagaTimeoutOperatorRequestOutcome.NotPending; + } + else if (string.IsNullOrEmpty(timeout.ManagementAddress)) + { + outcome = SagaTimeoutOperatorRequestOutcome.NoManagementAddress; + } + else if (timeout.PendingOperation != null && timeout.PendingOperation.RequestedAt > now - SagaTimeoutRules.OperatorRequestTimeout) + { + outcome = SagaTimeoutOperatorRequestOutcome.OperationPending; + } + else if (timeout.Version != request.ExpectedVersion) + { + outcome = SagaTimeoutOperatorRequestOutcome.VersionMismatch; + } + + if (outcome == SagaTimeoutOperatorRequestOutcome.Accepted) + { + timeout.PendingOperation = new PendingOperation + { + OperationId = request.OperationId, + Change = request.Change, + RequestedAt = request.RequestedAt.UtcDateTime, + RequestedDueTime = request.RequestedDueTime?.UtcDateTime + }; + await session.SaveChangesAsync(cancellationToken); + } + + return new SagaTimeoutOperatorRequestResult(outcome, ToView(timeout, now)); + } + + public async Task Forget(string timeoutId, CancellationToken cancellationToken = default) + { + using var session = await sessionProvider.OpenSession(cancellationToken: cancellationToken); + session.Delete(SagaTimeout.MakeDocumentId(timeoutId)); + await session.SaveChangesAsync(cancellationToken); + } + + static SagaTimeoutView ToView(SagaTimeout timeout, DateTime now) => new() + { + TimeoutId = timeout.TimeoutId, + EndpointName = timeout.EndpointName, + SagaType = timeout.SagaType, + SagaId = timeout.SagaId, + CorrelationValue = timeout.CorrelationValue, + TimeoutType = timeout.TimeoutType, + Version = timeout.Version, + TriggerId = timeout.TriggerId, + DueTime = timeout.DueTime, + Status = timeout.Status, + IsOverdue = SagaTimeoutRules.IsOverdue(timeout.Status, timeout.DueTime, now), + RequestedAt = timeout.RequestedAt, + LastChangeKind = timeout.LastChangeKind, + LastChangeSource = timeout.LastChangeSource, + LastChangedAt = timeout.LastChangedAt, + Host = timeout.Host, + ManagementAddress = timeout.ManagementAddress, + PendingOperation = ToView(timeout.PendingOperation) + }; + + static SagaTimeoutPendingOperationView ToView(PendingOperation operation) => operation == null + ? null + : new SagaTimeoutPendingOperationView + { + OperationId = operation.OperationId, + Change = operation.Change, + RequestedAt = operation.RequestedAt, + RequestedDueTime = operation.RequestedDueTime + }; + } + + static class SagaTimeoutsQueryExtensions + { + public static IRavenQueryable FilterByStatus(this IRavenQueryable query, SagaTimeoutListStatus status, DateTime overdueCutoff) => status switch + { + SagaTimeoutListStatus.Overdue => query.Where(timeout => timeout.Status == SagaTimeoutStatus.Pending && timeout.DueTime < overdueCutoff), + SagaTimeoutListStatus.All => query.Where(timeout => timeout.Status != SagaTimeoutStatus.Unknown), + SagaTimeoutListStatus.Pending => query.Where(timeout => timeout.Status == SagaTimeoutStatus.Pending && timeout.DueTime >= overdueCutoff), + _ => throw new ArgumentOutOfRangeException(nameof(status), status, null) + }; + } +} diff --git a/src/ServiceControl.Persistence.RavenDB/SagaTimeouts/SagaTimeoutsIndex.cs b/src/ServiceControl.Persistence.RavenDB/SagaTimeouts/SagaTimeoutsIndex.cs new file mode 100644 index 0000000000..24cc8da3aa --- /dev/null +++ b/src/ServiceControl.Persistence.RavenDB/SagaTimeouts/SagaTimeoutsIndex.cs @@ -0,0 +1,43 @@ +namespace ServiceControl.Persistence.RavenDB.SagaTimeouts +{ + using System; + using System.Linq; + using Raven.Client.Documents.Indexes; + using ServiceControl.Persistence.SagaTimeouts; + + class SagaTimeoutsIndex : AbstractIndexCreationTask + { + public SagaTimeoutsIndex() + { + Map = timeouts => + from timeout in timeouts + select new Result + { + Status = timeout.Status, + DueTime = timeout.DueTime, + EndpointName = timeout.EndpointName, + SagaType = timeout.SagaType, + SagaId = timeout.SagaId, + CorrelationValue = timeout.CorrelationValue, + TimeoutType = timeout.TimeoutType, + LastChangedAt = timeout.LastChangedAt, + Query = new[] { timeout.CorrelationValue, timeout.SagaType, timeout.TimeoutType, timeout.EndpointName } + }; + + Index(x => x.Query, FieldIndexing.Search); + } + + public class Result + { + public SagaTimeoutStatus Status { get; set; } + public DateTime? DueTime { get; set; } + public string EndpointName { get; set; } + public string SagaType { get; set; } + public Guid SagaId { get; set; } + public string CorrelationValue { get; set; } + public string TimeoutType { get; set; } + public DateTime? LastChangedAt { get; set; } + public string[] Query { get; set; } + } + } +} diff --git a/src/ServiceControl.Persistence.Tests.PostgreSql/ServiceControl.Persistence.Tests.PostgreSql.csproj b/src/ServiceControl.Persistence.Tests.PostgreSql/ServiceControl.Persistence.Tests.PostgreSql.csproj index 184b878a82..dab15d7ca3 100644 --- a/src/ServiceControl.Persistence.Tests.PostgreSql/ServiceControl.Persistence.Tests.PostgreSql.csproj +++ b/src/ServiceControl.Persistence.Tests.PostgreSql/ServiceControl.Persistence.Tests.PostgreSql.csproj @@ -31,6 +31,9 @@ + + + diff --git a/src/ServiceControl.Persistence.Tests.RavenDB/ApprovalFiles/RavenPersistedTypes.Verify.approved.txt b/src/ServiceControl.Persistence.Tests.RavenDB/ApprovalFiles/RavenPersistedTypes.Verify.approved.txt index 88bca6dca0..0fc4e6cb34 100644 --- a/src/ServiceControl.Persistence.Tests.RavenDB/ApprovalFiles/RavenPersistedTypes.Verify.approved.txt +++ b/src/ServiceControl.Persistence.Tests.RavenDB/ApprovalFiles/RavenPersistedTypes.Verify.approved.txt @@ -7,6 +7,8 @@ ServiceControl.Persistence.KnownEndpoint, ServiceControl.Persistence, Version=6. ServiceControl.Persistence.MessagesViewIndex+SortAndFilterOptions, ServiceControl.Persistence.RavenDB, Version=6.0.0.0, Culture=neutral, PublicKeyToken=null ServiceControl.Persistence.RavenDB.FailedMessageRetry, ServiceControl.Persistence.RavenDB, Version=6.0.0.0, Culture=neutral, PublicKeyToken=null ServiceControl.Persistence.RavenDB.RetryBatch, ServiceControl.Persistence.RavenDB, Version=6.0.0.0, Culture=neutral, PublicKeyToken=null +ServiceControl.Persistence.RavenDB.SagaTimeouts.SagaTimeout, ServiceControl.Persistence.RavenDB, Version=6.0.0.0, Culture=neutral, PublicKeyToken=null +ServiceControl.Persistence.RavenDB.SagaTimeouts.SagaTimeoutsIndex+Result, ServiceControl.Persistence.RavenDB, Version=6.0.0.0, Culture=neutral, PublicKeyToken=null ServiceControl.Persistence.RavenDB.Throughput.Models.EndpointDocument, ServiceControl.Persistence.RavenDB, Version=6.0.0.0, Culture=neutral, PublicKeyToken=null ServiceControl.Persistence.RetryBatchGroup, ServiceControl.Persistence, Version=6.0.0.0, Culture=neutral, PublicKeyToken=null ServiceControl.Recoverability.FailureGroupMessageView, ServiceControl.Persistence, Version=6.0.0.0, Culture=neutral, PublicKeyToken=null diff --git a/src/ServiceControl.Persistence.Tests.SqlServer/ServiceControl.Persistence.Tests.SqlServer.csproj b/src/ServiceControl.Persistence.Tests.SqlServer/ServiceControl.Persistence.Tests.SqlServer.csproj index d9c54a88e9..e09b040d4e 100644 --- a/src/ServiceControl.Persistence.Tests.SqlServer/ServiceControl.Persistence.Tests.SqlServer.csproj +++ b/src/ServiceControl.Persistence.Tests.SqlServer/ServiceControl.Persistence.Tests.SqlServer.csproj @@ -30,7 +30,9 @@ - + + + diff --git a/src/ServiceControl.Persistence.Tests/SagaTimeouts/SagaTimeoutDataStoreTests.cs b/src/ServiceControl.Persistence.Tests/SagaTimeouts/SagaTimeoutDataStoreTests.cs new file mode 100644 index 0000000000..1bf7924e86 --- /dev/null +++ b/src/ServiceControl.Persistence.Tests/SagaTimeouts/SagaTimeoutDataStoreTests.cs @@ -0,0 +1,275 @@ +namespace ServiceControl.Persistence.Tests.SagaTimeouts; + +using System; +using System.Linq; +using System.Text; +using System.Threading.Tasks; +using Microsoft.Extensions.DependencyInjection; +using NUnit.Framework; +using ServiceControl.Persistence.Infrastructure; +using ServiceControl.Persistence.SagaTimeouts; + +[TestFixture] +class SagaTimeoutDataStoreTests : PersistenceTestBase +{ + ISagaTimeoutDataStore Store => ServiceProvider.GetRequiredService(); + + DateTimeOffset Clock => new(Now); + + readonly Guid sagaId = Guid.NewGuid(); + readonly string timeoutId = Guid.NewGuid().ToString(); + + [Test] + public async Task Should_create_a_pending_timeout_from_a_request_and_ignore_its_duplicate() + { + var requested = Change(SagaTimeoutChangeKind.Requested, 1, dueIn: TimeSpan.FromHours(1)); + + Assert.That(await Store.Apply(requested), Is.EqualTo(SagaTimeoutApplyOutcome.Applied)); + Assert.That(await Store.Apply(requested), Is.EqualTo(SagaTimeoutApplyOutcome.Duplicate)); + + var details = await Store.Get(timeoutId); + + Assert.Multiple(() => + { + Assert.That(details.Status, Is.EqualTo(SagaTimeoutStatus.Pending)); + Assert.That(details.Version, Is.EqualTo(1)); + Assert.That(details.LastChangeKind, Is.EqualTo(SagaTimeoutChangeKind.Requested)); + Assert.That(details.LastChangeSource, Is.EqualTo(SagaTimeoutChangeSource.Saga)); + Assert.That(details.RequestedAt, Is.Not.Null); + Assert.That(details.ManagementAddress, Is.EqualTo("Sales.SagaTimeouts")); + Assert.That(details.Payload, Is.EqualTo("{\"Version\":1}")); + Assert.That(details.PayloadContentType, Is.EqualTo("application/json")); + Assert.That(details.History, Has.Count.EqualTo(1)); + }); + } + + [Test] + public async Task Should_keep_the_newest_version_when_an_older_change_arrives_late() + { + await Store.Apply(Change(SagaTimeoutChangeKind.Rescheduled, 3, dueIn: TimeSpan.FromHours(3))); + + var outcome = await Store.Apply(Change(SagaTimeoutChangeKind.Requested, 1, dueIn: TimeSpan.FromHours(1))); + + var details = await Store.Get(timeoutId); + + Assert.Multiple(() => + { + Assert.That(outcome, Is.EqualTo(SagaTimeoutApplyOutcome.HistoryOnly)); + Assert.That(details.Version, Is.EqualTo(3)); + Assert.That(details.LastChangeKind, Is.EqualTo(SagaTimeoutChangeKind.Rescheduled)); + Assert.That(details.Payload, Is.EqualTo("{\"Version\":3}")); + Assert.That(details.RequestedAt, Is.Not.Null, "The late request still tells when the timeout was requested"); + Assert.That(details.History.Select(entry => entry.Kind), Is.EquivalentTo(new[] { SagaTimeoutChangeKind.Requested, SagaTimeoutChangeKind.Rescheduled })); + }); + } + + [Test] + public async Task Should_let_a_cancel_win_over_a_reschedule_of_the_same_version() + { + await Store.Apply(Change(SagaTimeoutChangeKind.Requested, 1, dueIn: TimeSpan.FromHours(1))); + await Store.Apply(Change(SagaTimeoutChangeKind.Canceled, 2)); + await Store.Apply(Change(SagaTimeoutChangeKind.Rescheduled, 2, dueIn: TimeSpan.FromHours(2))); + + var details = await Store.Get(timeoutId); + + Assert.Multiple(() => + { + Assert.That(details.Status, Is.EqualTo(SagaTimeoutStatus.Canceled)); + Assert.That(details.History, Has.Count.EqualTo(3)); + }); + } + + [Test] + public async Task Should_list_pending_overdue_and_finished_timeouts_by_status() + { + var overdueId = Guid.NewGuid().ToString(); + var triggeredId = Guid.NewGuid().ToString(); + + await Store.Apply(Change(SagaTimeoutChangeKind.Requested, 1, dueIn: TimeSpan.FromHours(1))); + await Store.Apply(Change(SagaTimeoutChangeKind.Requested, 1, dueIn: TimeSpan.FromMinutes(-10)) with { TimeoutId = overdueId }); + await Store.Apply(Change(SagaTimeoutChangeKind.Requested, 1, dueIn: TimeSpan.FromMinutes(-1)) with { TimeoutId = triggeredId }); + await Store.Apply(Change(SagaTimeoutChangeKind.Triggered, 1) with { TimeoutId = triggeredId }); + await CompleteDatabaseOperation(); + + var pending = await QueryIds(SagaTimeoutListStatus.Pending); + var overdue = await QueryIds(SagaTimeoutListStatus.Overdue); + var all = await QueryIds(SagaTimeoutListStatus.All); + var counts = await Store.GetCounts(); + + Assert.Multiple(() => + { + Assert.That(pending, Is.EqualTo(new[] { timeoutId })); + Assert.That(overdue, Is.EqualTo(new[] { overdueId })); + Assert.That(all, Is.EquivalentTo(new[] { timeoutId, overdueId, triggeredId })); + Assert.That(counts.Pending, Is.EqualTo(1)); + Assert.That(counts.Overdue, Is.EqualTo(1)); + }); + } + + [Test] + public async Task Should_remove_the_timeouts_of_a_completed_saga_and_ignore_later_reports() + { + await Store.Apply(Change(SagaTimeoutChangeKind.Requested, 1, dueIn: TimeSpan.FromHours(1))); + + var completed = await Store.ApplySagaCompleted(new SagaCompletedChange("Sales", "Sales.OrderSaga", sagaId, Clock, Clock)); + var late = await Store.Apply(Change(SagaTimeoutChangeKind.Requested, 1, dueIn: TimeSpan.FromHours(1)) with { TimeoutId = Guid.NewGuid().ToString() }); + await CompleteDatabaseOperation(); + + var details = await Store.Get(timeoutId); + var sagaTimeouts = await Store.GetBySaga(sagaId); + + Assert.Multiple(() => + { + Assert.That(completed, Is.EqualTo(SagaTimeoutApplyOutcome.Applied)); + Assert.That(late, Is.EqualTo(SagaTimeoutApplyOutcome.Ignored)); + Assert.That(details, Is.Null); + Assert.That(sagaTimeouts, Is.Empty); + }); + } + + [Test] + public async Task Should_keep_a_rejection_for_an_unknown_timeout_out_of_the_lists() + { + var rejection = Rejection(Guid.NewGuid(), "TimeoutNotFound"); + + Assert.That(await Store.Apply(rejection), Is.EqualTo(SagaTimeoutApplyOutcome.HistoryOnly)); + Assert.That(await Store.Apply(rejection), Is.EqualTo(SagaTimeoutApplyOutcome.Duplicate)); + await CompleteDatabaseOperation(); + + var details = await Store.Get(timeoutId); + var all = await QueryIds(SagaTimeoutListStatus.All); + var sagaTimeouts = await Store.GetBySaga(sagaId); + + Assert.Multiple(() => + { + Assert.That(details.Status, Is.EqualTo(SagaTimeoutStatus.Unknown)); + Assert.That(details.History.Single().Reason, Is.EqualTo("TimeoutNotFound")); + Assert.That(all, Is.Empty); + Assert.That(sagaTimeouts, Is.Empty); + }); + } + + [Test] + public async Task Should_record_an_operator_request_until_the_endpoint_reports_it() + { + await Store.Apply(Change(SagaTimeoutChangeKind.Requested, 1, dueIn: TimeSpan.FromHours(1))); + + var operationId = Guid.NewGuid(); + var accepted = await Store.RecordOperatorRequest(Request(operationId, expectedVersion: 1)); + var whilePending = await Store.RecordOperatorRequest(Request(Guid.NewGuid(), expectedVersion: 1)); + + var pendingOperation = (await Store.Get(timeoutId)).PendingOperation; + + await Store.Apply(Change(SagaTimeoutChangeKind.Rescheduled, 2, dueIn: TimeSpan.FromHours(2)) with { OperationId = operationId }); + + var details = await Store.Get(timeoutId); + var staleVersion = await Store.RecordOperatorRequest(Request(Guid.NewGuid(), expectedVersion: 1)); + + Assert.Multiple(() => + { + Assert.That(accepted.Outcome, Is.EqualTo(SagaTimeoutOperatorRequestOutcome.Accepted)); + Assert.That(accepted.Timeout.ManagementAddress, Is.EqualTo("Sales.SagaTimeouts")); + Assert.That(whilePending.Outcome, Is.EqualTo(SagaTimeoutOperatorRequestOutcome.OperationPending)); + Assert.That(pendingOperation.OperationId, Is.EqualTo(operationId)); + Assert.That(details.PendingOperation, Is.Null); + Assert.That(details.LastChangeSource, Is.EqualTo(SagaTimeoutChangeSource.Operator)); + Assert.That(staleVersion.Outcome, Is.EqualTo(SagaTimeoutOperatorRequestOutcome.VersionMismatch)); + }); + } + + [Test] + public async Task Should_clear_the_pending_operation_when_the_endpoint_rejects_it() + { + await Store.Apply(Change(SagaTimeoutChangeKind.Requested, 1, dueIn: TimeSpan.FromHours(1))); + + var operationId = Guid.NewGuid(); + await Store.RecordOperatorRequest(Request(operationId, expectedVersion: 1)); + await Store.Apply(Rejection(operationId, "VersionMismatch")); + + var details = await Store.Get(timeoutId); + + Assert.Multiple(() => + { + Assert.That(details.PendingOperation, Is.Null); + Assert.That(details.Status, Is.EqualTo(SagaTimeoutStatus.Pending)); + Assert.That(details.History.Last().Kind, Is.EqualTo(SagaTimeoutChangeKind.Rejected)); + }); + } + + [Test] + public async Task Should_refuse_operator_requests_for_finished_or_missing_timeouts() + { + await Store.Apply(Change(SagaTimeoutChangeKind.Requested, 1, dueIn: TimeSpan.FromHours(1))); + await Store.Apply(Change(SagaTimeoutChangeKind.Triggered, 1)); + + var finished = await Store.RecordOperatorRequest(Request(Guid.NewGuid(), expectedVersion: 1)); + var missing = await Store.RecordOperatorRequest(Request(Guid.NewGuid(), expectedVersion: 1) with { TimeoutId = "missing" }); + + Assert.Multiple(() => + { + Assert.That(finished.Outcome, Is.EqualTo(SagaTimeoutOperatorRequestOutcome.NotPending)); + Assert.That(missing.Outcome, Is.EqualTo(SagaTimeoutOperatorRequestOutcome.NotFound)); + }); + } + + [Test] + public async Task Should_forget_a_timeout() + { + await Store.Apply(Change(SagaTimeoutChangeKind.Requested, 1, dueIn: TimeSpan.FromHours(1))); + + await Store.Forget(timeoutId); + + Assert.That(await Store.Get(timeoutId), Is.Null); + } + + async Task QueryIds(SagaTimeoutListStatus status) + { + var result = await Store.Query(new SagaTimeoutQuery { Status = status, SagaId = sagaId }, new PagingInfo(), new SortInfo("due_time", "asc")); + return result.Results.Select(timeout => timeout.TimeoutId).ToArray(); + } + + SagaTimeoutChange Change(SagaTimeoutChangeKind kind, int version, TimeSpan? dueIn = null) + { + var hasPayload = kind is SagaTimeoutChangeKind.Requested or SagaTimeoutChangeKind.Rescheduled; + + return new SagaTimeoutChange + { + Kind = kind, + TimeoutId = timeoutId, + TimeoutType = "Sales.PaymentTimeout", + Version = version, + TriggerId = kind == SagaTimeoutChangeKind.Canceled ? null : $"trigger-{version}", + DueTime = dueIn.HasValue ? Clock + dueIn.Value : null, + PayloadContentType = hasPayload ? "application/json" : null, + Payload = hasPayload ? Encoding.UTF8.GetBytes($"{{\"Version\":{version}}}") : null, + EndpointName = "Sales", + SagaType = "Sales.OrderSaga", + SagaId = sagaId, + CorrelationValue = "order-1", + ChangedAt = Clock, + ReceivedAt = Clock, + Host = "host", + HostId = "host-id", + ManagementAddress = "Sales.SagaTimeouts" + }; + } + + SagaTimeoutChange Rejection(Guid operationId, string reason) => new() + { + Kind = SagaTimeoutChangeKind.Rejected, + TimeoutId = timeoutId, + OperationId = operationId, + RequestedChange = "Reschedule", + RequestedDueTime = Clock, + Reason = reason, + EndpointName = "Sales", + SagaType = "Sales.OrderSaga", + SagaId = sagaId, + ChangedAt = Clock, + ReceivedAt = Clock, + ManagementAddress = "Sales.SagaTimeouts" + }; + + SagaTimeoutOperatorRequest Request(Guid operationId, int expectedVersion) => + new(timeoutId, operationId, SagaTimeoutOperation.Reschedule, expectedVersion, Clock + TimeSpan.FromHours(2), Clock); +} diff --git a/src/ServiceControl.Persistence/SagaTimeouts/ISagaTimeoutDataStore.cs b/src/ServiceControl.Persistence/SagaTimeouts/ISagaTimeoutDataStore.cs new file mode 100644 index 0000000000..751100ad31 --- /dev/null +++ b/src/ServiceControl.Persistence/SagaTimeouts/ISagaTimeoutDataStore.cs @@ -0,0 +1,47 @@ +#nullable enable +namespace ServiceControl.Persistence.SagaTimeouts; + +using System; +using System.Collections.Generic; +using System.Threading; +using System.Threading.Tasks; +using Infrastructure; + +/// +/// Keeps the projection of the saga timeouts that endpoints report. See docs/saga-timeout-management-design.md. +/// Every method owns its session, commits before returning, and returns detached snapshots. +/// +public interface ISagaTimeoutDataStore +{ + /// + /// Records one reported change. The current state moves forward only, ordered by (Version, Rank). + /// Returns when the saga has completed. + /// + Task Apply(SagaTimeoutChange change, CancellationToken cancellationToken = default); + + /// + /// Removes all timeouts of the saga and writes the completed-saga marker that makes later reports for the saga ignored. + /// + Task ApplySagaCompleted(SagaCompletedChange change, CancellationToken cancellationToken = default); + + Task>> Query(SagaTimeoutQuery query, PagingInfo paging, SortInfo sort, CancellationToken cancellationToken = default); + + Task GetCounts(CancellationToken cancellationToken = default); + + Task Get(string timeoutId, CancellationToken cancellationToken = default); + + /// + /// Returns the timeouts of one saga instance, except ones. + /// + Task> GetBySaga(Guid sagaId, CancellationToken cancellationToken = default); + + /// + /// Checks the expected version and stores the pending operation. Expected conflicts are returned as outcomes, not thrown. + /// + Task RecordOperatorRequest(SagaTimeoutOperatorRequest request, CancellationToken cancellationToken = default); + + /// + /// Removes the timeout and its history from ServiceControl. It does not change the saga. + /// + Task Forget(string timeoutId, CancellationToken cancellationToken = default); +} diff --git a/src/ServiceControl.Persistence/SagaTimeouts/SagaTimeoutModel.cs b/src/ServiceControl.Persistence/SagaTimeouts/SagaTimeoutModel.cs new file mode 100644 index 0000000000..8746b95115 --- /dev/null +++ b/src/ServiceControl.Persistence/SagaTimeouts/SagaTimeoutModel.cs @@ -0,0 +1,155 @@ +#nullable enable +namespace ServiceControl.Persistence.SagaTimeouts; + +using System; + +public enum SagaTimeoutStatus +{ + Unknown, + Pending, + Triggered, + Canceled +} + +public enum SagaTimeoutChangeKind +{ + Requested, + Rescheduled, + Canceled, + Triggered, + Rejected +} + +public enum SagaTimeoutChangeSource +{ + Saga, + Operator +} + +public enum SagaTimeoutOperation +{ + Reschedule, + Cancel, + Trigger +} + +public enum SagaTimeoutApplyOutcome +{ + /// The change became the current state. + Applied, + /// The change was older than the current state, or is a rejection, so only its history entry was recorded. + HistoryOnly, + /// The history already has an entry with the same deduplication key. + Duplicate, + /// The saga has completed, so the report was dropped. + Ignored +} + +public static class SagaTimeoutRules +{ + /// Absorbs delayed-delivery latency and clock drift before a pending timeout counts as overdue. + public static readonly TimeSpan OverdueGracePeriod = TimeSpan.FromMinutes(5); + + /// A pending operator request older than this no longer blocks a new one. + public static readonly TimeSpan OperatorRequestTimeout = TimeSpan.FromMinutes(10); + + public const int MaxHistoryEntries = 100; + + public static SagaTimeoutStatus StatusAfter(SagaTimeoutChangeKind kind) => kind switch + { + SagaTimeoutChangeKind.Requested or SagaTimeoutChangeKind.Rescheduled => SagaTimeoutStatus.Pending, + SagaTimeoutChangeKind.Triggered => SagaTimeoutStatus.Triggered, + SagaTimeoutChangeKind.Canceled => SagaTimeoutStatus.Canceled, + SagaTimeoutChangeKind.Rejected => throw new ArgumentOutOfRangeException(nameof(kind), kind, "A rejection does not change the state."), + _ => throw new ArgumentOutOfRangeException(nameof(kind), kind, null) + }; + + // Terminal states win a tie on Version, because Canceled does not increment it + public static int RankOf(SagaTimeoutChangeKind kind) => kind is SagaTimeoutChangeKind.Triggered or SagaTimeoutChangeKind.Canceled ? 1 : 0; + + public static string DeduplicationKey(SagaTimeoutChange change) => change.Kind == SagaTimeoutChangeKind.Rejected + ? $"Rejected:{change.RequestedChange}:{change.ChangedAt.UtcTicks}" + : $"{change.Kind}:{change.Version}"; + + public static bool IsOverdue(SagaTimeoutStatus status, DateTime? dueTime, DateTime now) => + status == SagaTimeoutStatus.Pending && dueTime < now - OverdueGracePeriod; +} + +/// +/// One change reported by an endpoint. comes from the endpoint clock and from ServiceControl. +/// +public sealed record SagaTimeoutChange +{ + public required SagaTimeoutChangeKind Kind { get; init; } + public required string TimeoutId { get; init; } + public required string EndpointName { get; init; } + public required string SagaType { get; init; } + public required Guid SagaId { get; init; } + public string? CorrelationValue { get; init; } + + // Not reported by a rejection + public string? TimeoutType { get; init; } + public int Version { get; init; } + + public string? TriggerId { get; init; } + public DateTimeOffset? DueTime { get; init; } + public string? PayloadContentType { get; init; } + public byte[]? Payload { get; init; } + + public Guid? OperationId { get; init; } + + public string? RequestedChange { get; init; } + public DateTimeOffset? RequestedDueTime { get; init; } + public string? Reason { get; init; } + + public required DateTimeOffset ChangedAt { get; init; } + public required DateTimeOffset ReceivedAt { get; init; } + public string? Host { get; init; } + public string? HostId { get; init; } + public string? ManagementAddress { get; init; } + + public SagaTimeoutChangeSource Source => OperationId.HasValue ? SagaTimeoutChangeSource.Operator : SagaTimeoutChangeSource.Saga; +} + +public sealed record SagaCompletedChange(string EndpointName, string SagaType, Guid SagaId, DateTimeOffset ChangedAt, DateTimeOffset ReceivedAt); + +public enum SagaTimeoutListStatus +{ + /// Pending and not overdue. + Pending, + Overdue, + /// Pending and finished timeouts, without the Unknown ones. + All +} + +public sealed record SagaTimeoutQuery +{ + public SagaTimeoutListStatus Status { get; init; } = SagaTimeoutListStatus.Pending; + public string? EndpointName { get; init; } + public string? SagaType { get; init; } + public Guid? SagaId { get; init; } + public string? SearchText { get; init; } + + public static readonly string[] SortOptions = ["due_time", "last_changed_at", "endpoint", "saga_type", "timeout_type"]; +} + +public sealed record SagaTimeoutOperatorRequest( + string TimeoutId, + Guid OperationId, + SagaTimeoutOperation Change, + int ExpectedVersion, + DateTimeOffset? RequestedDueTime, + DateTimeOffset RequestedAt); + +public enum SagaTimeoutOperatorRequestOutcome +{ + Accepted, + NotFound, + VersionMismatch, + OperationPending, + NotPending, + NoManagementAddress +} + +/// The timeout after the request was recorded; null when it does not exist. +public sealed record SagaTimeoutOperatorRequestResult(SagaTimeoutOperatorRequestOutcome Outcome, SagaTimeoutView? Timeout = null); diff --git a/src/ServiceControl.Persistence/SagaTimeouts/SagaTimeoutView.cs b/src/ServiceControl.Persistence/SagaTimeouts/SagaTimeoutView.cs new file mode 100644 index 0000000000..71caa897a7 --- /dev/null +++ b/src/ServiceControl.Persistence/SagaTimeouts/SagaTimeoutView.cs @@ -0,0 +1,90 @@ +#nullable enable +namespace ServiceControl.Persistence.SagaTimeouts; + +using System; +using System.Collections.Generic; +using System.Text.Json.Serialization; + +// The API writes these enum values as their names ("Pending"), unlike the camel-case default of other APIs, +// because ServicePulse shares this contract. +public class SagaTimeoutView +{ + public required string TimeoutId { get; init; } + public required string EndpointName { get; init; } + public required string SagaType { get; init; } + public Guid SagaId { get; init; } + public string? CorrelationValue { get; init; } + public string? TimeoutType { get; init; } + public int Version { get; init; } + public string? TriggerId { get; init; } + public DateTime? DueTime { get; init; } + + [JsonConverter(typeof(JsonStringEnumConverter))] + public SagaTimeoutStatus Status { get; init; } + + public bool IsOverdue { get; init; } + public DateTime? RequestedAt { get; init; } + + [JsonConverter(typeof(JsonStringEnumConverter))] + public SagaTimeoutChangeKind LastChangeKind { get; init; } + + [JsonConverter(typeof(JsonStringEnumConverter))] + public SagaTimeoutChangeSource LastChangeSource { get; init; } + + public DateTime? LastChangedAt { get; init; } + public string? Host { get; init; } + public string? ManagementAddress { get; init; } + + // Written as null rather than left out, so clients can rely on the field + [JsonIgnore(Condition = JsonIgnoreCondition.Never)] + public SagaTimeoutPendingOperationView? PendingOperation { get; init; } +} + +public class SagaTimeoutPendingOperationView +{ + public Guid OperationId { get; init; } + + [JsonConverter(typeof(JsonStringEnumConverter))] + public SagaTimeoutOperation Change { get; init; } + + public DateTime RequestedAt { get; init; } + public DateTime? RequestedDueTime { get; init; } +} + +public class SagaTimeoutDetails : SagaTimeoutView +{ + public string? PayloadContentType { get; init; } + + /// The current payload decoded as UTF-8. + public string? Payload { get; init; } + + public bool HistoryTruncated { get; init; } + + public IList History { get; init; } = []; +} + +public class SagaTimeoutHistoryEntryView +{ + [JsonConverter(typeof(JsonStringEnumConverter))] + public SagaTimeoutChangeKind Kind { get; init; } + + public int Version { get; init; } + public string? TriggerId { get; init; } + public DateTime? DueTime { get; init; } + public DateTime ChangedAt { get; init; } + public DateTime ReceivedAt { get; init; } + + [JsonConverter(typeof(JsonStringEnumConverter))] + public SagaTimeoutChangeSource Source { get; init; } + + public Guid? OperationId { get; init; } + public string? RequestedChange { get; init; } + public DateTime? RequestedDueTime { get; init; } + public string? Reason { get; init; } +} + +public class SagaTimeoutCounts +{ + public long Pending { get; init; } + public long Overdue { get; init; } +} diff --git a/src/ServiceControl.UnitTests/ApprovalFiles/APIApprovals.HttpApiRoutes.approved.txt b/src/ServiceControl.UnitTests/ApprovalFiles/APIApprovals.HttpApiRoutes.approved.txt index d9f20d7486..fb7d1ed282 100644 --- a/src/ServiceControl.UnitTests/ApprovalFiles/APIApprovals.HttpApiRoutes.approved.txt +++ b/src/ServiceControl.UnitTests/ApprovalFiles/APIApprovals.HttpApiRoutes.approved.txt @@ -78,3 +78,11 @@ PUT /redirects/{messageRedirectId:guid} => ServiceControl.MessageRedirects.Api.M POST /retention/purge => ServiceControl.Retention.Api.SystemMaintenanceController:Purge(RetentionPurgeRequest request, CancellationToken cancellationToken) GET /retention/purge/status => ServiceControl.Retention.Api.SystemMaintenanceController:Status(CancellationToken cancellationToken) GET /sagas/{id} => ServiceControl.SagaAudit.SagasController:Sagas(PagingInfo pagingInfo, Guid id, CancellationToken cancellationToken) +GET /sagas/{sagaId}/timeouts => ServiceControl.SagaTimeouts.Web.SagaTimeoutsController:SagaTimeouts(Guid sagaId, CancellationToken cancellationToken) +GET /timeouts => ServiceControl.SagaTimeouts.Web.SagaTimeoutsController:Timeouts(PagingInfo pagingInfo, String status, String endpoint, String sagaType, Nullable sagaId, String q, String sort, String direction, CancellationToken cancellationToken) +DELETE /timeouts/{timeoutId} => ServiceControl.SagaTimeouts.Web.SagaTimeoutsController:Forget(String timeoutId, CancellationToken cancellationToken) +GET /timeouts/{timeoutId} => ServiceControl.SagaTimeouts.Web.SagaTimeoutsController:Timeout(String timeoutId, CancellationToken cancellationToken) +POST /timeouts/{timeoutId}/cancel => ServiceControl.SagaTimeouts.Web.SagaTimeoutsController:Cancel(String timeoutId, SagaTimeoutVersionRequest request, CancellationToken cancellationToken) +POST /timeouts/{timeoutId}/reschedule => ServiceControl.SagaTimeouts.Web.SagaTimeoutsController:Reschedule(String timeoutId, RescheduleSagaTimeoutRequest request, CancellationToken cancellationToken) +POST /timeouts/{timeoutId}/trigger => ServiceControl.SagaTimeouts.Web.SagaTimeoutsController:Trigger(String timeoutId, SagaTimeoutVersionRequest request, CancellationToken cancellationToken) +GET /timeouts/counts => ServiceControl.SagaTimeouts.Web.SagaTimeoutsController:Counts(CancellationToken cancellationToken) diff --git a/src/ServiceControl.UnitTests/SagaTimeouts/SagaTimeoutManagementRequestTests.cs b/src/ServiceControl.UnitTests/SagaTimeouts/SagaTimeoutManagementRequestTests.cs new file mode 100644 index 0000000000..0ad09e6994 --- /dev/null +++ b/src/ServiceControl.UnitTests/SagaTimeouts/SagaTimeoutManagementRequestTests.cs @@ -0,0 +1,96 @@ +namespace ServiceControl.UnitTests.SagaTimeouts +{ + using System; + using System.Text; + using System.Text.Json; + using NServiceBus; + using NUnit.Framework; + using ServiceControl.Persistence.SagaTimeouts; + using ServiceControl.SagaTimeouts; + using ManagementHeaders = ServiceControl.SagaTimeouts.RequestSagaTimeoutChangeHandler.ManagementHeaders; + + [TestFixture] + class SagaTimeoutManagementRequestTests + { + [Test] + public void Should_send_a_trigger_as_a_reschedule_with_an_empty_body() + { + var dueTime = new DateTimeOffset(2026, 10, 19, 12, 30, 15, TimeSpan.FromHours(2)).AddTicks(1234560); + + var request = RequestSagaTimeoutChangeHandler.BuildRequest(Change(SagaTimeoutOperation.Trigger, dueTime)); + + Assert.Multiple(() => + { + Assert.That(request.Headers[ManagementHeaders.Change], Is.EqualTo("Reschedule")); + Assert.That(request.Headers[ManagementHeaders.DueTime], Is.EqualTo("2026-10-19 10:30:15:123456 Z")); + Assert.That(request.Headers[ManagementHeaders.ExpectedVersion], Is.EqualTo("3")); + Assert.That(request.Headers[ManagementHeaders.SagaType], Is.EqualTo("Sales.OrderSaga")); + Assert.That(request.Headers[ManagementHeaders.TimeoutId], Is.EqualTo("timeout-1")); + Assert.That(request.Headers[Headers.MessageId], Is.EqualTo(request.MessageId)); + Assert.That(request.Headers.ContainsKey(Headers.ContentType), Is.False); + Assert.That(request.Body.IsEmpty, Is.True); + }); + } + + [Test] + public void Should_send_a_new_payload_as_the_body_of_a_reschedule() + { + var change = Change(SagaTimeoutOperation.Reschedule, DateTimeOffset.UtcNow); + change.Payload = "{\"Amount\":1}"; + change.PayloadContentType = "application/json"; + + var request = RequestSagaTimeoutChangeHandler.BuildRequest(change); + + Assert.Multiple(() => + { + Assert.That(Encoding.UTF8.GetString(request.Body.Span), Is.EqualTo("{\"Amount\":1}")); + Assert.That(request.Headers[Headers.ContentType], Is.EqualTo("application/json")); + }); + } + + [Test] + public void Should_send_a_cancel_without_a_due_time() + { + var request = RequestSagaTimeoutChangeHandler.BuildRequest(Change(SagaTimeoutOperation.Cancel, null)); + + Assert.Multiple(() => + { + Assert.That(request.Headers[ManagementHeaders.Change], Is.EqualTo("Cancel")); + Assert.That(request.Headers.ContainsKey(ManagementHeaders.DueTime), Is.False); + Assert.That(request.Body.IsEmpty, Is.True); + }); + } + + [Test] + public void Should_read_a_report_as_the_endpoint_serializes_it() + { + const string json = """ + {"TriggerId":"t1","DueTime":"2026-10-19T10:30:00+00:00","PayloadContentType":"application/json","Payload":"eyJBIjoxfQ==", + "TimeoutId":"timeout-1","TimeoutType":"Sales.PaymentTimeout","Version":1,"EndpointName":"Sales","SagaType":"Sales.OrderSaga", + "SagaId":"5c1b2a3e-1111-4c4c-8888-000000000001","CorrelationPropertyValue":"order-1","ChangedAt":"2026-10-19T10:00:00+00:00", + "Host":"host","HostId":"host-id","ManagementAddress":"Sales.SagaTimeouts"} + """; + + var message = JsonSerializer.Deserialize(json, SagaTimeoutMessagesSerializationContext.Default.SagaTimeoutRequested); + + Assert.Multiple(() => + { + Assert.That(Encoding.UTF8.GetString(message.Payload), Is.EqualTo("{\"A\":1}")); + Assert.That(message.ManagementAddress, Is.EqualTo("Sales.SagaTimeouts")); + Assert.That(message.Version, Is.EqualTo(1)); + }); + } + + static RequestSagaTimeoutChange Change(SagaTimeoutOperation operation, DateTimeOffset? dueTime) => new() + { + OperationId = Guid.NewGuid(), + Change = operation, + ManagementAddress = "Sales.SagaTimeouts", + SagaType = "Sales.OrderSaga", + SagaId = Guid.NewGuid(), + TimeoutId = "timeout-1", + ExpectedVersion = 3, + DueTime = dueTime + }; + } +} diff --git a/src/ServiceControl/Infrastructure/NServiceBusFactory.cs b/src/ServiceControl/Infrastructure/NServiceBusFactory.cs index fce530c4d8..306b6bac71 100644 --- a/src/ServiceControl/Infrastructure/NServiceBusFactory.cs +++ b/src/ServiceControl/Infrastructure/NServiceBusFactory.cs @@ -68,6 +68,7 @@ public static void Configure(Settings.Settings settings, ITransportCustomization SagaAuditMessagesSerializationContext.Default, HeartbeatSerializationContext.Default, UsageReportingSerializationContext.Default, + SagaTimeoutMessagesSerializationContext.Default, // This is required until we move all known message types over to source generated contexts new DefaultJsonTypeInfoResolver() } diff --git a/src/ServiceControl/SagaTimeouts/RequestSagaTimeoutChange.cs b/src/ServiceControl/SagaTimeouts/RequestSagaTimeoutChange.cs new file mode 100644 index 0000000000..18b037b0f3 --- /dev/null +++ b/src/ServiceControl/SagaTimeouts/RequestSagaTimeoutChange.cs @@ -0,0 +1,21 @@ +namespace ServiceControl.SagaTimeouts +{ + using System; + using NServiceBus; + using ServiceControl.Persistence.SagaTimeouts; + + // Carries everything the satellite request needs, so the handler does not read the store + class RequestSagaTimeoutChange : ICommand + { + public Guid OperationId { get; set; } + public SagaTimeoutOperation Change { get; set; } + public string ManagementAddress { get; set; } + public string SagaType { get; set; } + public Guid SagaId { get; set; } + public string TimeoutId { get; set; } + public int ExpectedVersion { get; set; } + public DateTimeOffset? DueTime { get; set; } + public string Payload { get; set; } + public string PayloadContentType { get; set; } + } +} diff --git a/src/ServiceControl/SagaTimeouts/RequestSagaTimeoutChangeHandler.cs b/src/ServiceControl/SagaTimeouts/RequestSagaTimeoutChangeHandler.cs new file mode 100644 index 0000000000..231ce90a27 --- /dev/null +++ b/src/ServiceControl/SagaTimeouts/RequestSagaTimeoutChangeHandler.cs @@ -0,0 +1,89 @@ +namespace ServiceControl.SagaTimeouts +{ + using System; + using System.Collections.Generic; + using System.Globalization; + using System.Text; + using System.Threading.Tasks; + using Infrastructure.Auth; + using NServiceBus; + using NServiceBus.Routing; + using NServiceBus.Transport; + using ServiceControl.Persistence.SagaTimeouts; + + [Handler] + class RequestSagaTimeoutChangeHandler(IMessageDispatcher dispatcher, IMessageActionAuditLog auditLog) : IHandleMessages + { + public async Task Handle(RequestSagaTimeoutChange message, IMessageHandlerContext context) + { + var request = BuildRequest(message); + var transportTransaction = context.Extensions.GetOrCreate(); + + await dispatcher.Dispatch( + new TransportOperations(new TransportOperation(request, new UnicastAddressTag(message.ManagementAddress))), + transportTransaction, + context.CancellationToken); + + var (user, operationId) = AuditHeaders.Read(context.MessageHeaders); + if (!string.IsNullOrEmpty(operationId)) + { + auditLog.MessageAction(user, ToAuditKind(message.Change), Permissions.ErrorTimeoutsManage, MessageActionScope.Single, message.TimeoutId, operationId); + } + } + + // The satellite does not deserialize the body, so every field of the request travels in a header + public static OutgoingMessage BuildRequest(RequestSagaTimeoutChange message) + { + var messageId = Guid.CreateVersion7().ToString("D"); + var isCancel = message.Change == SagaTimeoutOperation.Cancel; + + var headers = new Dictionary + { + [Headers.MessageId] = messageId, + [ManagementHeaders.OperationId] = message.OperationId.ToString("D"), + [ManagementHeaders.SagaType] = message.SagaType, + [ManagementHeaders.SagaId] = message.SagaId.ToString("D"), + [ManagementHeaders.TimeoutId] = message.TimeoutId, + [ManagementHeaders.ExpectedVersion] = message.ExpectedVersion.ToString(CultureInfo.InvariantCulture), + [ManagementHeaders.Change] = isCancel ? ManagementHeaders.CancelChange : ManagementHeaders.RescheduleChange + }; + + if (!isCancel) + { + var dueTime = message.DueTime ?? throw new InvalidOperationException($"A {message.Change} request for timeout {message.TimeoutId} has no due time."); + headers[ManagementHeaders.DueTime] = DateTimeOffsetHelper.ToWireFormattedString(dueTime); + } + + var body = Array.Empty(); + if (message.Change == SagaTimeoutOperation.Reschedule && message.Payload != null) + { + body = Encoding.UTF8.GetBytes(message.Payload); + headers[Headers.ContentType] = message.PayloadContentType ?? "application/json"; + } + + return new OutgoingMessage(messageId, headers, body); + } + + static MessageActionKind ToAuditKind(SagaTimeoutOperation change) => change switch + { + SagaTimeoutOperation.Cancel => MessageActionKind.Cancel, + SagaTimeoutOperation.Trigger => MessageActionKind.Trigger, + SagaTimeoutOperation.Reschedule => MessageActionKind.Reschedule, + _ => throw new ArgumentOutOfRangeException(nameof(change), change, null) + }; + + public static class ManagementHeaders + { + public const string OperationId = "NServiceBus.SagaTimeoutManagement.OperationId"; + public const string SagaType = "NServiceBus.SagaTimeoutManagement.SagaType"; + public const string SagaId = "NServiceBus.SagaTimeoutManagement.SagaId"; + public const string TimeoutId = "NServiceBus.SagaTimeoutManagement.TimeoutId"; + public const string ExpectedVersion = "NServiceBus.SagaTimeoutManagement.ExpectedVersion"; + public const string Change = "NServiceBus.SagaTimeoutManagement.Change"; + public const string DueTime = "NServiceBus.SagaTimeoutManagement.DueTime"; + + public const string RescheduleChange = "Reschedule"; + public const string CancelChange = "Cancel"; + } + } +} diff --git a/src/ServiceControl/SagaTimeouts/SagaTimeoutPlatformMessages.cs b/src/ServiceControl/SagaTimeouts/SagaTimeoutPlatformMessages.cs new file mode 100644 index 0000000000..5253373e73 --- /dev/null +++ b/src/ServiceControl/SagaTimeouts/SagaTimeoutPlatformMessages.cs @@ -0,0 +1,72 @@ +#nullable enable +// Endpoints report saga timeout changes with these contracts. The namespace and type names must match the +// endpoint's copies, because NServiceBus.EnclosedMessageTypes carries only the full type name. +namespace NServiceBus; + +using System; +using System.Text.Json.Serialization; + +abstract class SagaTimeoutPlatformMessage : IMessage +{ + public required string EndpointName { get; set; } + public required string SagaType { get; set; } + public required Guid SagaId { get; set; } + public string? CorrelationPropertyValue { get; set; } + public required DateTimeOffset ChangedAt { get; set; } + public required string Host { get; set; } + public required string HostId { get; set; } + public required string ManagementAddress { get; set; } +} + +abstract class SagaTimeoutChangeMessage : SagaTimeoutPlatformMessage +{ + public required string TimeoutId { get; set; } + public required string TimeoutType { get; set; } + public required int Version { get; set; } +} + +sealed class SagaTimeoutRequested : SagaTimeoutChangeMessage +{ + public required string TriggerId { get; set; } + public required DateTimeOffset DueTime { get; set; } + public required string PayloadContentType { get; set; } + public required byte[] Payload { get; set; } +} + +sealed class SagaTimeoutRescheduled : SagaTimeoutChangeMessage +{ + public Guid? OperationId { get; set; } + public required string TriggerId { get; set; } + public required DateTimeOffset DueTime { get; set; } + public required string PayloadContentType { get; set; } + public required byte[] Payload { get; set; } +} + +sealed class SagaTimeoutCanceled : SagaTimeoutChangeMessage +{ + public Guid? OperationId { get; set; } +} + +sealed class SagaTimeoutTriggered : SagaTimeoutChangeMessage +{ + public required string TriggerId { get; set; } +} + +sealed class SagaTimeoutChangeRejected : SagaTimeoutPlatformMessage +{ + public Guid? OperationId { get; set; } + public required string TimeoutId { get; set; } + public required string RequestedChange { get; set; } + public DateTimeOffset? RequestedDueTime { get; set; } + public required string Reason { get; set; } +} + +sealed class SagaCompleted : SagaTimeoutPlatformMessage; + +[JsonSerializable(typeof(SagaTimeoutRequested))] +[JsonSerializable(typeof(SagaTimeoutRescheduled))] +[JsonSerializable(typeof(SagaTimeoutCanceled))] +[JsonSerializable(typeof(SagaTimeoutTriggered))] +[JsonSerializable(typeof(SagaTimeoutChangeRejected))] +[JsonSerializable(typeof(SagaCompleted))] +partial class SagaTimeoutMessagesSerializationContext : JsonSerializerContext; diff --git a/src/ServiceControl/SagaTimeouts/SagaTimeoutReportHandler.cs b/src/ServiceControl/SagaTimeouts/SagaTimeoutReportHandler.cs new file mode 100644 index 0000000000..d3bf5e701f --- /dev/null +++ b/src/ServiceControl/SagaTimeouts/SagaTimeoutReportHandler.cs @@ -0,0 +1,47 @@ +namespace ServiceControl.SagaTimeouts +{ + using System.Threading.Tasks; + using NServiceBus; + using ServiceControl.Persistence.SagaTimeouts; + + [Handler] + class SagaTimeoutReportHandler(SagaTimeoutReportProcessor processor) : + IHandleMessages, + IHandleMessages, + IHandleMessages, + IHandleMessages, + IHandleMessages, + IHandleMessages + { + public Task Handle(SagaTimeoutRequested message, IMessageHandlerContext context) => + processor.Apply(message, SagaTimeoutChangeKind.Requested, change => change with + { + TriggerId = message.TriggerId, + DueTime = message.DueTime, + PayloadContentType = message.PayloadContentType, + Payload = message.Payload + }, context.CancellationToken); + + public Task Handle(SagaTimeoutRescheduled message, IMessageHandlerContext context) => + processor.Apply(message, SagaTimeoutChangeKind.Rescheduled, change => change with + { + TriggerId = message.TriggerId, + DueTime = message.DueTime, + PayloadContentType = message.PayloadContentType, + Payload = message.Payload, + OperationId = message.OperationId + }, context.CancellationToken); + + public Task Handle(SagaTimeoutCanceled message, IMessageHandlerContext context) => + processor.Apply(message, SagaTimeoutChangeKind.Canceled, change => change with { OperationId = message.OperationId }, context.CancellationToken); + + public Task Handle(SagaTimeoutTriggered message, IMessageHandlerContext context) => + processor.Apply(message, SagaTimeoutChangeKind.Triggered, change => change with { TriggerId = message.TriggerId }, context.CancellationToken); + + public Task Handle(SagaTimeoutChangeRejected message, IMessageHandlerContext context) => + processor.ApplyRejection(message, context.CancellationToken); + + public Task Handle(SagaCompleted message, IMessageHandlerContext context) => + processor.ApplySagaCompleted(message, context.CancellationToken); + } +} diff --git a/src/ServiceControl/SagaTimeouts/SagaTimeoutReportProcessor.cs b/src/ServiceControl/SagaTimeouts/SagaTimeoutReportProcessor.cs new file mode 100644 index 0000000000..8e7bfe91d0 --- /dev/null +++ b/src/ServiceControl/SagaTimeouts/SagaTimeoutReportProcessor.cs @@ -0,0 +1,70 @@ +namespace ServiceControl.SagaTimeouts +{ + using System; + using System.Threading; + using System.Threading.Tasks; + using Microsoft.Extensions.Logging; + using NServiceBus; + using ServiceControl.Persistence.SagaTimeouts; + + class SagaTimeoutReportProcessor(ISagaTimeoutDataStore store, TimeProvider timeProvider, ILogger logger) + { + public async Task Apply(SagaTimeoutChangeMessage message, SagaTimeoutChangeKind kind, Func addKindFields, CancellationToken cancellationToken = default) + { + var change = addKindFields(new SagaTimeoutChange + { + Kind = kind, + TimeoutId = message.TimeoutId, + TimeoutType = message.TimeoutType, + Version = message.Version, + EndpointName = message.EndpointName, + SagaType = message.SagaType, + SagaId = message.SagaId, + CorrelationValue = message.CorrelationPropertyValue, + ChangedAt = message.ChangedAt, + ReceivedAt = timeProvider.GetUtcNow(), + Host = message.Host, + HostId = message.HostId, + ManagementAddress = message.ManagementAddress + }); + + await Apply(change, cancellationToken); + } + + public Task ApplyRejection(SagaTimeoutChangeRejected message, CancellationToken cancellationToken = default) => + Apply(new SagaTimeoutChange + { + Kind = SagaTimeoutChangeKind.Rejected, + TimeoutId = message.TimeoutId, + EndpointName = message.EndpointName, + SagaType = message.SagaType, + SagaId = message.SagaId, + CorrelationValue = message.CorrelationPropertyValue, + OperationId = message.OperationId, + RequestedChange = message.RequestedChange, + RequestedDueTime = message.RequestedDueTime, + Reason = message.Reason, + ChangedAt = message.ChangedAt, + ReceivedAt = timeProvider.GetUtcNow(), + Host = message.Host, + HostId = message.HostId, + ManagementAddress = message.ManagementAddress + }, cancellationToken); + + public async Task ApplySagaCompleted(SagaCompleted message, CancellationToken cancellationToken = default) + { + var outcome = await store.ApplySagaCompleted( + new SagaCompletedChange(message.EndpointName, message.SagaType, message.SagaId, message.ChangedAt, timeProvider.GetUtcNow()), + cancellationToken); + + logger.LogDebug("Saga {SagaType} {SagaId} completed: {Outcome}", message.SagaType, message.SagaId, outcome); + } + + async Task Apply(SagaTimeoutChange change, CancellationToken cancellationToken) + { + var outcome = await store.Apply(change, cancellationToken); + + logger.LogDebug("Saga timeout {TimeoutId} {Kind} version {Version}: {Outcome}", change.TimeoutId, change.Kind, change.Version, outcome); + } + } +} diff --git a/src/ServiceControl/SagaTimeouts/SagaTimeoutsComponent.cs b/src/ServiceControl/SagaTimeouts/SagaTimeoutsComponent.cs new file mode 100644 index 0000000000..7d7c3a111a --- /dev/null +++ b/src/ServiceControl/SagaTimeouts/SagaTimeoutsComponent.cs @@ -0,0 +1,15 @@ +namespace ServiceControl.SagaTimeouts +{ + using Microsoft.Extensions.DependencyInjection; + using Microsoft.Extensions.Hosting; + using Particular.ServiceControl; + using ServiceBus.Management.Infrastructure.Settings; + using Transports; + + // See docs/saga-timeout-management-design.md. The data store comes from the persister. + class SagaTimeoutsComponent : ServiceControlComponent + { + public override void Configure(Settings settings, ITransportCustomization transportCustomization, IHostApplicationBuilder hostBuilder) => + hostBuilder.Services.AddSingleton(); + } +} diff --git a/src/ServiceControl/SagaTimeouts/Web/SagaTimeoutsController.cs b/src/ServiceControl/SagaTimeouts/Web/SagaTimeoutsController.cs new file mode 100644 index 0000000000..7cc1ae1e4f --- /dev/null +++ b/src/ServiceControl/SagaTimeouts/Web/SagaTimeoutsController.cs @@ -0,0 +1,219 @@ +namespace ServiceControl.SagaTimeouts.Web +{ + using System; + using System.Collections.Generic; + using System.Threading; + using System.Threading.Tasks; + using Infrastructure.Auth; + using Infrastructure.WebApi; + using Microsoft.AspNetCore.Authorization; + using Microsoft.AspNetCore.Mvc; + using NServiceBus; + using Persistence.Infrastructure; + using Persistence.SagaTimeouts; + + [ApiController] + [Route("api")] + public class SagaTimeoutsController( + ISagaTimeoutDataStore store, + IMessageSession session, + TimeProvider timeProvider, + ICurrentUserAccessor userAccessor, + IMessageActionAuditLog auditLog) : ControllerBase + { + [Authorize(Policy = Permissions.ErrorTimeoutsView)] + [Route("timeouts")] + [HttpGet] + public async Task>> Timeouts( + [FromQuery] PagingInfo pagingInfo, + [FromQuery] string status = null, + [FromQuery] string endpoint = null, + [FromQuery(Name = "saga_type")] string sagaType = null, + [FromQuery(Name = "saga_id")] Guid? sagaId = null, + [FromQuery] string q = null, + [FromQuery] string sort = null, + [FromQuery] string direction = null, + CancellationToken cancellationToken = default) + { + if (!TryParseStatus(status, out var listStatus)) + { + return BadRequest(new SagaTimeoutActionError($"Unknown status '{status}'. Use pending, overdue, or all.")); + } + + var query = new SagaTimeoutQuery + { + Status = listStatus, + EndpointName = endpoint, + SagaType = sagaType, + SagaId = sagaId, + SearchText = q + }; + + var result = await store.Query(query, pagingInfo, ToSortInfo(sort, direction), cancellationToken); + + Response.WithQueryStatsAndPagingInfo(result.QueryStats, pagingInfo); + + return Ok(result.Results); + } + + [Authorize(Policy = Permissions.ErrorTimeoutsView)] + [Route("timeouts/counts")] + [HttpGet] + public Task Counts(CancellationToken cancellationToken = default) => store.GetCounts(cancellationToken); + + [Authorize(Policy = Permissions.ErrorTimeoutsView)] + [Route("timeouts/{timeoutId}")] + [HttpGet] + public async Task> Timeout(string timeoutId, CancellationToken cancellationToken = default) + { + var details = await store.Get(timeoutId, cancellationToken); + + return details == null ? NotFound() : Ok(details); + } + + [Authorize(Policy = Permissions.ErrorTimeoutsView)] + [Route("sagas/{sagaId}/timeouts")] + [HttpGet] + public Task> SagaTimeouts(Guid sagaId, CancellationToken cancellationToken = default) => store.GetBySaga(sagaId, cancellationToken); + + [Authorize(Policy = Permissions.ErrorTimeoutsManage)] + [Route("timeouts/{timeoutId}/reschedule")] + [HttpPost] + public Task Reschedule(string timeoutId, [FromBody] RescheduleSagaTimeoutRequest request, CancellationToken cancellationToken = default) + { + if (request.DueTime == null) + { + return Task.FromResult(BadRequest(new SagaTimeoutActionError("A reschedule needs a due_time."))); + } + + return RequestChange(timeoutId, SagaTimeoutOperation.Reschedule, request.ExpectedVersion, request.DueTime, request.Payload, request.PayloadContentType, cancellationToken); + } + + [Authorize(Policy = Permissions.ErrorTimeoutsManage)] + [Route("timeouts/{timeoutId}/trigger")] + [HttpPost] + public Task Trigger(string timeoutId, [FromBody] SagaTimeoutVersionRequest request, CancellationToken cancellationToken = default) => + RequestChange(timeoutId, SagaTimeoutOperation.Trigger, request.ExpectedVersion, timeProvider.GetUtcNow(), null, null, cancellationToken); + + [Authorize(Policy = Permissions.ErrorTimeoutsManage)] + [Route("timeouts/{timeoutId}/cancel")] + [HttpPost] + public Task Cancel(string timeoutId, [FromBody] SagaTimeoutVersionRequest request, CancellationToken cancellationToken = default) => + RequestChange(timeoutId, SagaTimeoutOperation.Cancel, request.ExpectedVersion, null, null, null, cancellationToken); + + [Authorize(Policy = Permissions.ErrorTimeoutsManage)] + [Route("timeouts/{timeoutId}")] + [HttpDelete] + public async Task Forget(string timeoutId, CancellationToken cancellationToken = default) + { + var user = userAccessor.Resolve(User); + + await auditLog.AuditedOperation(user, MessageActionKind.Delete, Permissions.ErrorTimeoutsManage, MessageActionScope.Single, + resource: timeoutId, count: 1, operationId: this.AuditOperationId(), + ct => store.Forget(timeoutId, ct), cancellationToken); + + return NoContent(); + } + + async Task RequestChange(string timeoutId, SagaTimeoutOperation change, int? expectedVersion, DateTimeOffset? dueTime, string payload, string payloadContentType, CancellationToken cancellationToken) + { + if (expectedVersion == null) + { + return BadRequest(new SagaTimeoutActionError("The request needs an expected_version.")); + } + + var operationId = Guid.NewGuid(); + var result = await store.RecordOperatorRequest( + new SagaTimeoutOperatorRequest(timeoutId, operationId, change, expectedVersion.Value, dueTime, timeProvider.GetUtcNow()), + cancellationToken); + + switch (result.Outcome) + { + case SagaTimeoutOperatorRequestOutcome.NotFound: + return NotFound(); + case SagaTimeoutOperatorRequestOutcome.VersionMismatch: + return Conflict(new SagaTimeoutActionError($"The timeout is at version {result.Timeout.Version}, not {expectedVersion}.")); + case SagaTimeoutOperatorRequestOutcome.OperationPending: + return Conflict(new SagaTimeoutActionError("Another change of this timeout is waiting for the endpoint.")); + case SagaTimeoutOperatorRequestOutcome.NotPending: + return UnprocessableEntity(new SagaTimeoutActionError($"The timeout is {result.Timeout.Status}, so it cannot be changed.")); + case SagaTimeoutOperatorRequestOutcome.NoManagementAddress: + return UnprocessableEntity(new SagaTimeoutActionError("The endpoint has not reported the address that accepts timeout changes.")); + case SagaTimeoutOperatorRequestOutcome.Accepted: + break; + default: + throw new InvalidOperationException($"Unknown operator request outcome: {result.Outcome}"); + } + + var timeout = result.Timeout; + var user = userAccessor.Resolve(User); + var auditOperationId = this.AuditOperationId(); + + await auditLog.AuditedOperation(user, ToAuditKind(change), Permissions.ErrorTimeoutsManage, MessageActionScope.Single, + resource: timeoutId, count: 1, operationId: auditOperationId, + ct => session.Send(new RequestSagaTimeoutChange + { + OperationId = operationId, + Change = change, + ManagementAddress = timeout.ManagementAddress, + SagaType = timeout.SagaType, + SagaId = timeout.SagaId, + TimeoutId = timeout.TimeoutId, + ExpectedVersion = expectedVersion.Value, + DueTime = dueTime, + Payload = payload, + PayloadContentType = payloadContentType + }, AuditHeaders.LocalSendOptions(user, auditOperationId), ct), cancellationToken); + + return Accepted(new SagaTimeoutOperationAccepted(operationId)); + } + + static MessageActionKind ToAuditKind(SagaTimeoutOperation change) => change switch + { + SagaTimeoutOperation.Cancel => MessageActionKind.Cancel, + SagaTimeoutOperation.Trigger => MessageActionKind.Trigger, + SagaTimeoutOperation.Reschedule => MessageActionKind.Reschedule, + _ => throw new ArgumentOutOfRangeException(nameof(change), change, null) + }; + + static bool TryParseStatus(string status, out SagaTimeoutListStatus listStatus) + { + listStatus = SagaTimeoutListStatus.Pending; + switch (status) + { + case null or "" or "pending": + return true; + case "overdue": + listStatus = SagaTimeoutListStatus.Overdue; + return true; + case "all": + listStatus = SagaTimeoutListStatus.All; + return true; + default: + return false; + } + } + + // Timeouts have their own sort options and sort ascending by default, so the shared SortInfo binder does not fit + static SortInfo ToSortInfo(string sort, string direction) => new( + Array.IndexOf(SagaTimeoutQuery.SortOptions, sort) >= 0 ? sort : "due_time", + direction == "desc" ? "desc" : "asc"); + } + + public class RescheduleSagaTimeoutRequest + { + public DateTimeOffset? DueTime { get; set; } + public int? ExpectedVersion { get; set; } + public string Payload { get; set; } + public string PayloadContentType { get; set; } + } + + public class SagaTimeoutVersionRequest + { + public int? ExpectedVersion { get; set; } + } + + public record SagaTimeoutOperationAccepted(Guid OperationId); + + public record SagaTimeoutActionError(string Detail); +} diff --git a/src/ServiceControl/ServiceControlMainInstance.cs b/src/ServiceControl/ServiceControlMainInstance.cs index 10a7ba1a0b..1f49e04a96 100644 --- a/src/ServiceControl/ServiceControlMainInstance.cs +++ b/src/ServiceControl/ServiceControlMainInstance.cs @@ -5,6 +5,7 @@ namespace Particular.ServiceControl using global::ServiceControl.ExternalIntegrations; using global::ServiceControl.Monitoring; using global::ServiceControl.Recoverability; + using global::ServiceControl.SagaTimeouts; static class ServiceControlMainInstance { @@ -15,6 +16,7 @@ static class ServiceControlMainInstance new RecoverabilityComponent(), new HeartbeatMonitoringComponent(), new CustomChecksComponent(), + new SagaTimeoutsComponent(), new LicensingComponent() }; }