diff --git a/docs/site/src/content/docs/grains/reminders/dynamodb.md b/docs/site/src/content/docs/grains/reminders/dynamodb.md index 06e77825b15..3d3b0be22ca 100644 --- a/docs/site/src/content/docs/grains/reminders/dynamodb.md +++ b/docs/site/src/content/docs/grains/reminders/dynamodb.md @@ -1,7 +1,7 @@ --- title: Configure Amazon DynamoDB reminders description: Configure durable Orleans reminder storage with Amazon DynamoDB. -ms.date: 08/20/2026 +ms.date: 08/28/2026 ms.topic: how-to --- @@ -26,3 +26,54 @@ When and to `true` and configure and for provisioned capacity. and allow the provider to create the reminder table and update its provisioned capacity. Infrastructure-managed provisioning keeps table lifecycle and capacity changes in the deployment workflow. + +## Legacy-schema consistency hardening + +`TableMode=Legacy` retains the existing table and indexes while hardening reminder scheduling against GSI lag: + +- GSI range and grain results are discovery candidates. The provider strongly point-reads every candidate from the base table before returning it, so an obsolete index row cannot resurrect a deleted reminder or restore an old schedule. +- Before a silo removes a known local reminder which a range query omitted, it strongly point-reads that identity. A still-present or newer row is reconciled instead of removed. +- Reminder registration, update, and removal strongly point-read and reconcile before returning when the handling silo remains the owner. A stale handler sends a bounded, three-hop reconciliation notification through current ring views; the receiving owner strongly point-reads before applying it. Per-identity sequencing prevents out-of-order point reads from restoring an older schedule. +- Legacy mode never uses a full-table scan during startup, topology changes, or periodic refresh. All discovery remains cost-bounded to GSI queries and base-table point reads. + +This path materially hardens runtime scheduling without changing the key schema, but it cannot provide a complete strong set read. A GSI can omit a row, and a point read can validate only identities already known to the caller. Cold startup and newly acquired ranges therefore cannot cheaply discover an identity which is still absent from the GSI. A mutation notification which exhausts its topology hops, reaches a pre-protocol binary, or encounters a transport failure has the same boundary and is logged. Arbitrary `GetReminders` and range-table calls can also omit completed writes until GSI convergence. Retries do not change this limitation, and Legacy mode does not disguise it with scans. V2 remains required for the full point, grain, and range completed-write visibility guarantee. + +Legacy refresh cost is one eventual GSI query for each contiguous owned ring subrange, plus one strongly consistent base-table `GetItem` for each candidate and for each locally known identity omitted by the GSI. A wrap-around range uses two GSI queries. A stale-owner mutation adds at most four system-target messages and, if ownership changes after reads begin, up to four strong point reads. Startup and topology changes use the same bounded operations and consume no scan capacity. Full-table scans occur only inside the explicit V2 migration coordinator, never as the steady-state Legacy correctness mechanism. + +## Migrate to the strongly consistent schema + +The default is `Legacy`, so upgrading Orleans does not change an existing table. The V2 schema stores reminders in a separate table named by , or `${TableName}-v2` when that option is unset. Both names continue to support custom values. + +V2 uses 32 service-scoped hash buckets. Its base-table partition key combines an encoded `ServiceId` with the bucket selected by the unsigned grain hash. Its sort key contains the eight-digit uppercase hexadecimal grain hash followed by tagged, delimiter-safe grain-identity and reminder-name components. Short components use base64url; long components use a SHA-256 digest so the key remains below DynamoDB's 1,024-byte sort-key limit. Writes conditionally verify the complete stored identity, so a digest collision fails instead of overwriting another reminder. These fixed-width prefixes make `(begin, end]`, wrap-around, grain-prefix, and point queries lexicographically exact. Point, grain, and range reads use strongly consistent base-table operations. A range refresh issues bounded key queries against the 32 buckets instead of scanning the table. + +### Prerequisites + +- Enable point-in-time recovery or take an on-demand backup of the legacy table. +- Grant `CreateTable`, `DescribeTable`, `Scan`, `Query`, `GetItem`, `PutItem`, `DeleteItem`, and `TransactWriteItems` permissions for both table names. Infrastructure-managed deployments must provision the V2 table with string `PartitionKey` and string `SortKey` base keys before migration and set `CreateIfNotExists` and `UpdateIfExists` accordingly. +- Size V2 write capacity for the temporary dual-write period. Migration uses strongly consistent legacy scans and transactional copy operations. Set to bound each resumable scan page; a smaller value reduces bursts but increases requests. +- Keep DynamoDB transactions available. The two tables must remain in the same AWS account and Region. + +### Rolling protocol + +1. Deploy the V2-capable binary to every silo with `TableMode=Migrate`. Reads remain on V1. Each completed registration and removal writes V1 and V2 atomically with the same ETag. A leased coordinator strongly scans V1, conditionally copies each row, removes V2 rows whose V1 source is still absent, and verifies both schemas. Persisted page keys make interruption and lease takeover resumable and idempotent. +2. Wait for every silo to run the new binary and for the migration log to report `Ready`. Keep this stage for at least the longest expected old deployment shutdown and AWS SDK retry interval. +3. Change every silo to `TableMode=V2`. Finalization fails closed if any active Orleans membership entry lacks a fresh V2-capability marker. It repeats reconciliation and exact verification, confirms membership did not change, and only then persists `Cutover`. V2-capable silos consult this strongly consistent state before reads, so already-running `Migrate` silos follow the cutover while the configuration rollout completes. +4. Keep `TableMode=V2` and retain V1 for the desired rollback window. Dual writes continue after cutover. Do not run a binary which predates this protocol after cutover: old binaries do not understand the fence and can write only V1. The capability check covers active silos at transition time, not a future operator-initiated downgrade. + +This is a two-phase deployment, not a one-step rolling cutover. A direct `Legacy` to `V2` full-cluster restart is also safe when all old processes are stopped before the first V2 silo starts. Finalization never uses a GSI or a periodic full-table scan. + +### Roll back and retire V1 + +Set every V2-capable silo to `TableMode=Rollback`. The coordinator requires compatible active silos, reconciles and verifies the transactionally maintained copies, then persists `RolledBack`; all V2-capable silos return to V1 reads. After correcting the issue, repeat `Migrate` and `V2`. + +The provider never deletes or disables the legacy table. After the rollback window, set every silo to `TableMode=V2Only`. The coordinator performs one last fenced verification with stable compatible membership and persists the irreversible `Retired` state. Already-running V2-capable silos observe that state before each mutation and stop writing V1. After every silo reports `Retired`, archive or delete V1 through the deployment system. `V2Only` startup does not recreate V1, and rollback is no longer available. + +### States, recovery, and operations + +Migration metadata is isolated by `ServiceId` in V2. `Backfilling` stores the last evaluated V1 key after each completed page; replay after a crash is harmless because the source ETag is checked transactionally. `Verifying` performs final reconciliation. `Ready` means exact source and target counts and contents matched. `Cutover` selects V2 reads. `RolledBack` selects V1 reads. `VerificationFailed` preserves V1 reads and prevents cutover. `Retired` selects V2-only reads and writes and is irreversible. + +Only one unexpired migration lease can advance a service. Other `Migrate` silos continue serving V1 and dual-writing. A stopped owner is recoverable after lease expiry; restarting any configured silo resumes from the checkpoint. Conditional conflicts caused by an old V1 writer replay the same page instead of advancing past the change. A V1 delete cannot resurrect a stale V2 row because copy and source-ETag validation are one transaction. + +Orleans logs state transitions, page numbers and matching item counts, lease contention, and verification failures. Monitor those logs together with DynamoDB `ConsumedReadCapacityUnits`, `ConsumedWriteCapacityUnits`, `ReadThrottleEvents`, `WriteThrottleEvents`, `TransactionConflict`, and `SystemErrors` metrics for both tables. `VerificationFailed`, repeated lease loss, or an incompatible-silo error requires operator action and never changes the read schema. + +Backfill scans are migration-only and strongly consistent. Because a filtered DynamoDB scan consumes capacity for every evaluated item, migrating one `ServiceId` in a large shared V1 table reads the full physical table. Schedule large migrations around capacity limits, reduce `MigrationPageSize`, and migrate services separately. This protocol does not depend on DynamoDB Streams retention; interruption recovery uses persisted scan keys plus a final full reconciliation. If an external change-capture process is used operationally, its stream-retention window does not replace the final verification. diff --git a/src/AWS/Orleans.Reminders.DynamoDB/DynamoDBRemindersProviderBuilder.cs b/src/AWS/Orleans.Reminders.DynamoDB/DynamoDBRemindersProviderBuilder.cs index 7692e73d83a..078ed12ed7b 100644 --- a/src/AWS/Orleans.Reminders.DynamoDB/DynamoDBRemindersProviderBuilder.cs +++ b/src/AWS/Orleans.Reminders.DynamoDB/DynamoDBRemindersProviderBuilder.cs @@ -2,6 +2,7 @@ using Orleans; using Orleans.Hosting; using Orleans.Providers; +using Orleans.Configuration; [assembly: RegisterProvider("DynamoDB", "Reminders", "Silo", typeof(DynamoDBRemindersProviderBuilder))] @@ -73,6 +74,22 @@ public void Configure(ISiloBuilder builder, string? name, IConfigurationSection { options.UpdateIfExists = uie; } + + var v2TableName = configurationSection[nameof(options.V2TableName)]; + if (!string.IsNullOrEmpty(v2TableName)) + { + options.V2TableName = v2TableName; + } + + if (Enum.TryParse(configurationSection[nameof(options.TableMode)], ignoreCase: true, out var tableMode)) + { + options.TableMode = tableMode; + } + + if (int.TryParse(configurationSection[nameof(options.MigrationPageSize)], out var migrationPageSize)) + { + options.MigrationPageSize = migrationPageSize; + } }); } } diff --git a/src/AWS/Orleans.Reminders.DynamoDB/README.md b/src/AWS/Orleans.Reminders.DynamoDB/README.md index e7fdacde94a..c6e5c70e393 100644 --- a/src/AWS/Orleans.Reminders.DynamoDB/README.md +++ b/src/AWS/Orleans.Reminders.DynamoDB/README.md @@ -51,6 +51,17 @@ await host.WaitForShutdownAsync(); `UseDynamoDBReminderService` configures reminder storage independently from the cluster membership provider. The AWS SDK credential and profile resolution chain supplies credentials when the reminder options omit explicit keys. This example uses on-demand capacity and an infrastructure-managed table. +## Strongly consistent schema migration + +The provider defaults to the legacy schema. Legacy reads strongly point-validate GSI candidates and locally known omissions, and completed mutations send a bounded owner notification followed by a point read. This prevents stale resurrection and removal but cannot make a GSI omission discoverable for cold startup, newly acquired ranges, exhausted notifications, or arbitrary set reads. Legacy startup and refresh never use full-table scans. + +V2 migration supplies the complete guarantee and is an explicit two-stage rollout: + +1. Deploy every silo with `TableMode=Migrate` to create/backfill `${TableName}-v2` while retaining V1 reads and transactional dual writes. +2. After all silos are upgraded and migration reports `Ready`, deploy `TableMode=V2`. Cutover verifies the copies and fails if any active silo lacks a V2 compatibility marker. + +V2 point, grain, and hash-range reads query the sharded base table with strong consistency. V1 remains transactionally maintained for rollback. After the rollback window, `TableMode=V2Only` performs an irreversible fenced transition before operators retire V1; the provider never deletes it. See [Configure Amazon DynamoDB reminders](https://dotnet.github.io/orleans/docs/grains/reminders/dynamodb/) for prerequisites, recovery, rollback, capacity, and compatibility details. + ## Example - Using Reminders in a Grain ```csharp using System; diff --git a/src/AWS/Orleans.Reminders.DynamoDB/Reminders/DynamoDBReminderTable.LegacyHardening.cs b/src/AWS/Orleans.Reminders.DynamoDB/Reminders/DynamoDBReminderTable.LegacyHardening.cs new file mode 100644 index 00000000000..b26a1ba18a2 --- /dev/null +++ b/src/AWS/Orleans.Reminders.DynamoDB/Reminders/DynamoDBReminderTable.LegacyHardening.cs @@ -0,0 +1,31 @@ +using Orleans.Runtime; + +namespace Orleans.Reminders.DynamoDB; + +internal sealed partial class DynamoDBReminderTable +{ + private const int LegacyPointReadConcurrency = 16; + + private async Task> ConfirmLegacyDiscoveryCandidates(List discovered) + { + IReadOnlyList candidates = testHooks?.LegacyDiscoveryResults?.Invoke(discovered) ?? discovered; + var result = new List(candidates.Count); + foreach (var batch in candidates.BatchIEnumerable(LegacyPointReadConcurrency)) + { + var reads = batch.Select(entry => ReadLegacyRow(entry.GrainId, entry.ReminderName)); + foreach (var entry in await Task.WhenAll(reads)) + { + if (entry is not null) + { + result.Add(entry); + } + } + } + + return result; + } + + private Task ReadLegacyRow(GrainId grainId, string reminderName) + => storage.ReadSingleEntryAsync(options.TableName, GetLegacyKey(grainId, reminderName), Resolve); + +} diff --git a/src/AWS/Orleans.Reminders.DynamoDB/Reminders/DynamoDBReminderTable.Migration.cs b/src/AWS/Orleans.Reminders.DynamoDB/Reminders/DynamoDBReminderTable.Migration.cs new file mode 100644 index 00000000000..fdb458529ae --- /dev/null +++ b/src/AWS/Orleans.Reminders.DynamoDB/Reminders/DynamoDBReminderTable.Migration.cs @@ -0,0 +1,884 @@ +using Amazon.DynamoDBv2.Model; +using Microsoft.Extensions.Logging; +using Orleans.Configuration; +using Orleans.Runtime; +using System.Globalization; + +namespace Orleans.Reminders.DynamoDB; + +internal sealed class DynamoDBReminderMigrationTestHooks +{ + public Func? AfterLegacyPageRead { get; init; } + + public Func? BeforeVerification { get; init; } + + public Func? AfterPageCheckpoint { get; init; } + + public Func, IReadOnlyList>? LegacyDiscoveryResults { get; init; } +} + +internal sealed partial class DynamoDBReminderTable +{ + private const string MigrationStateSortKey = "STATE"; + private const string MigrationLeaseSortKey = "LEASE"; + private const string NodeSortKeyPrefix = "NODE#"; + private const string StatusAttribute = "MigrationStatus"; + private const string OwnerAttribute = "LeaseOwner"; + private const string LeaseTokenAttribute = "LeaseToken"; + private const string ExpiresAtAttribute = "ExpiresAt"; + private const string CheckpointReminderIdAttribute = "CheckpointReminderId"; + private const string CheckpointGrainHashAttribute = "CheckpointGrainHash"; + private const string SourceCountAttribute = "SourceCount"; + private const string TargetCountAttribute = "TargetCount"; + private static readonly TimeSpan MigrationLeaseDuration = TimeSpan.FromMinutes(2); + private static readonly TimeSpan CompatibilityMarkerLifetime = TimeSpan.FromMinutes(2); + private static readonly TimeSpan CompatibilityHeartbeatPeriod = TimeSpan.FromSeconds(30); + + private void ValidateOptions() + { + if (options.MigrationPageSize <= 0) + { + throw new ArgumentOutOfRangeException(nameof(options.MigrationPageSize), "Migration page size must be positive."); + } + + if (string.IsNullOrWhiteSpace(options.TableName)) + { + throw new ArgumentException("The legacy reminder table name must not be empty.", nameof(options.TableName)); + } + + if (string.IsNullOrWhiteSpace(v2TableName) || v2TableName.Length > 255) + { + throw new ArgumentException("The V2 reminder table name must contain between 1 and 255 characters.", nameof(options.V2TableName)); + } + + if (string.Equals(options.TableName, v2TableName, StringComparison.Ordinal)) + { + throw new ArgumentException("The V2 reminder table must be different from the legacy table.", nameof(options.V2TableName)); + } + } + + private async Task RefreshReadMode() + { + if (options.TableMode == DynamoDBReminderTableMode.Legacy) + { + return; + } + + var state = await ReadMigrationState(); + useV2Reads = state?.Status is MigrationStatus.Cutover or MigrationStatus.Retired; + useV2OnlyWrites = state?.Status == MigrationStatus.Retired; + } + + private async Task InitializeMigration(CancellationToken cancellationToken) + { + var state = await ReadMigrationState(); + if (state?.Status == MigrationStatus.Retired) + { + if (options.TableMode == DynamoDBReminderTableMode.Rollback) + { + throw new InvalidOperationException("The DynamoDB reminder V1 schema has been retired and can no longer be rolled back."); + } + + await EnsureCompatibleCluster(cancellationToken); + useV2Reads = true; + useV2OnlyWrites = true; + return; + } + + if (state?.Status == MigrationStatus.Cutover && options.TableMode != DynamoDBReminderTableMode.Rollback) + { + if (options.TableMode == DynamoDBReminderTableMode.V2Only) + { + await RetireLegacySchema(cancellationToken); + return; + } + + await EnsureCompatibleCluster(cancellationToken); + useV2Reads = true; + return; + } + + if (options.TableMode == DynamoDBReminderTableMode.V2Only) + { + throw new InvalidOperationException("V2Only mode requires the service to be in the Cutover state."); + } + + var waitForLease = options.TableMode is DynamoDBReminderTableMode.V2 or DynamoDBReminderTableMode.Rollback; + if (!await AcquireMigrationLease(waitForLease, cancellationToken)) + { + useV2Reads = state?.Status == MigrationStatus.Cutover; + LogMigrationLeaseContended(logger, serviceId, migrationOwner); + return; + } + + state = await ReadMigrationState(); + if (state?.Status == MigrationStatus.Retired) + { + throw new InvalidOperationException("The DynamoDB reminder V1 schema was retired while waiting for the migration lease."); + } + + if (state?.Status == MigrationStatus.Cutover && options.TableMode == DynamoDBReminderTableMode.V2) + { + useV2Reads = true; + await ReleaseMigrationLease(); + return; + } + + StartLeaseRenewal(); + try + { + if (options.TableMode == DynamoDBReminderTableMode.Rollback) + { + await EnsureCompatibleCluster(cancellationToken); + await ReconcileAndVerify(cancellationToken, preserveMigrationState: true); + await WriteMigrationState(new(MigrationStatus.RolledBack)); + useV2Reads = false; + LogMigrationState(logger, serviceId, MigrationStatus.RolledBack.ToString()); + return; + } + + MembershipVersion? membershipVersion = null; + if (options.TableMode == DynamoDBReminderTableMode.V2) + { + membershipVersion = await EnsureCompatibleCluster(cancellationToken); + } + + await BackfillAndVerify(cancellationToken); + + if (options.TableMode == DynamoDBReminderTableMode.V2) + { + var finalMembershipVersion = await EnsureCompatibleCluster(cancellationToken); + if (finalMembershipVersion != membershipVersion) + { + throw new InvalidOperationException( + "Cluster membership changed during DynamoDB reminder finalization. " + + "Migration remains verified but was not cut over; retry after membership stabilizes."); + } + + await WriteMigrationState(new(MigrationStatus.Cutover)); + useV2Reads = true; + LogMigrationState(logger, serviceId, MigrationStatus.Cutover.ToString()); + } + else + { + useV2Reads = false; + } + } + finally + { + await StopLeaseRenewal(); + await ReleaseMigrationLease(); + } + } + + private async Task RetireLegacySchema(CancellationToken cancellationToken) + { + if (!await AcquireMigrationLease(wait: true, cancellationToken)) + { + throw new InvalidOperationException("Unable to acquire the DynamoDB reminder migration lease for V1 retirement."); + } + + StartLeaseRenewal(); + try + { + var state = await ReadMigrationState(); + if (state?.Status == MigrationStatus.Retired) + { + useV2Reads = true; + useV2OnlyWrites = true; + return; + } + + if (state?.Status != MigrationStatus.Cutover) + { + throw new InvalidOperationException($"V1 retirement requires Cutover state, but found '{state?.Status.ToString() ?? "missing"}'."); + } + + var membershipVersion = await EnsureCompatibleCluster(cancellationToken); + await ReconcileAndVerify(cancellationToken, preserveMigrationState: true); + var finalMembershipVersion = await EnsureCompatibleCluster(cancellationToken); + if (finalMembershipVersion != membershipVersion) + { + throw new InvalidOperationException( + "Cluster membership changed during DynamoDB reminder V1 retirement. Retry after membership stabilizes."); + } + + await WriteMigrationState(new(MigrationStatus.Retired)); + useV2Reads = true; + useV2OnlyWrites = true; + LogMigrationState(logger, serviceId, MigrationStatus.Retired.ToString()); + } + finally + { + await StopLeaseRenewal(); + await ReleaseMigrationLease(); + } + } + + private async Task BackfillAndVerify(CancellationToken cancellationToken) + { + var state = await ReadMigrationState() ?? new(MigrationStatus.Backfilling); + state.Status = MigrationStatus.Backfilling; + await WriteMigrationState(state); + LogMigrationState(logger, serviceId, state.Status.ToString()); + + var checkpoint = state.GetCheckpoint(); + var pageNumber = 0; + while (true) + { + cancellationToken.ThrowIfCancellationRequested(); + var (records, nextCheckpoint) = await storage.ScanPageAsync( + options.TableName, + new() { [":service"] = new(serviceId) }, + $"{SERVICE_ID_PROPERTY_NAME} = :service", + static item => new LegacyReminderRecord(item), + options.MigrationPageSize, + checkpoint, + cancellationToken); + + if (testHooks?.AfterLegacyPageRead is { } afterPageRead) + { + await afterPageRead(); + } + + var conflict = false; + foreach (var record in records) + { + if (!await CopyLegacyRecord(record)) + { + conflict = true; + break; + } + } + + if (conflict) + { + continue; + } + + checkpoint = nextCheckpoint; + state.SetCheckpoint(checkpoint); + await WriteMigrationState(state); + await RenewMigrationLease(); + LogMigrationPage(logger, serviceId, ++pageNumber, records.Count); + if (testHooks?.AfterPageCheckpoint is { } afterPageCheckpoint) + { + await afterPageCheckpoint(); + } + + if (checkpoint is not { Count: > 0 }) + { + break; + } + } + + await ReconcileAndVerify(cancellationToken, preserveMigrationState: false); + } + + private async Task CopyLegacyRecord(LegacyReminderRecord record) + { + var entry = Resolve(record.Item); + var values = new Dictionary { [":sourceETag"] = Clone(record.Item[ETAG_PROPERTY_NAME]) }; + try + { + await storage.WriteTxAsync( + [ + new() { ConditionCheck = CreateMigrationLeaseFence() }, + new() + { + ConditionCheck = new() + { + TableName = options.TableName, + Key = GetLegacyKey(entry.GrainId, entry.ReminderName), + ConditionExpression = $"{ETAG_PROPERTY_NAME} = :sourceETag", + ExpressionAttributeValues = values, + }, + }, + new() + { + Put = CreateIdentitySafeV2Put(CreateV2ItemFromLegacy(record.Item, entry), entry), + }, + ]); + return true; + } + catch (TransactionCanceledException exception) when (IsConditionalFailureAt(exception, 0)) + { + throw new InvalidOperationException("The DynamoDB reminder migration lease was lost while copying a row.", exception); + } + catch (TransactionCanceledException exception) when (IsConditionalFailureAt(exception, 1)) + { + return false; + } + } + + private async Task ReconcileAndVerify(CancellationToken cancellationToken, bool preserveMigrationState) + { + if (!preserveMigrationState) + { + var state = new MigrationState(MigrationStatus.Verifying); + await WriteMigrationState(state); + LogMigrationState(logger, serviceId, state.Status.ToString()); + } + + await ForEachLegacyRecord(async sourceRecord => + { + cancellationToken.ThrowIfCancellationRequested(); + if (!await CopyLegacyRecord(sourceRecord)) + { + throw new InvalidOperationException("A legacy reminder changed during final verification. Retry migration."); + } + }, cancellationToken); + + await ForEachV2Record(async targetRecord => + { + cancellationToken.ThrowIfCancellationRequested(); + var entry = Resolve(targetRecord.Item); + var sourceRecord = await storage.ReadSingleEntryAsync( + options.TableName, + GetLegacyKey(entry.GrainId, entry.ReminderName), + static item => new LegacyReminderRecord(item)); + if (sourceRecord is null) + { + await DeleteOrphan(targetRecord); + } + else if (!RecordsEqual(sourceRecord.Item, targetRecord.Item) && !await CopyLegacyRecord(sourceRecord)) + { + throw new InvalidOperationException("A legacy reminder changed during final reconciliation. Retry migration."); + } + }, cancellationToken); + + if (testHooks?.BeforeVerification is { } beforeVerification) + { + await beforeVerification(); + } + + var sourceCount = 0; + var targetCount = 0; + var verified = true; + await ForEachLegacyRecord(async sourceRecord => + { + sourceCount++; + var entry = Resolve(sourceRecord.Item); + var targetRecord = await storage.ReadSingleEntryAsync( + v2TableName, + GetV2Key(entry.GrainId, entry.ReminderName), + static item => new V2ReminderRecord(item)); + verified &= targetRecord is not null && RecordsEqual(sourceRecord.Item, targetRecord.Item); + }, cancellationToken); + await ForEachV2Record(async targetRecord => + { + targetCount++; + var entry = Resolve(targetRecord.Item); + var sourceRecord = await storage.ReadSingleEntryAsync( + options.TableName, + GetLegacyKey(entry.GrainId, entry.ReminderName), + static item => new LegacyReminderRecord(item)); + verified &= sourceRecord is not null && RecordsEqual(sourceRecord.Item, targetRecord.Item); + }, cancellationToken); + verified &= sourceCount == targetCount; + + if (!verified) + { + if (!preserveMigrationState) + { + await WriteMigrationState(new(MigrationStatus.VerificationFailed, sourceCount, targetCount)); + } + + LogMigrationVerificationFailed(logger, serviceId, sourceCount, targetCount); + throw new InvalidOperationException( + $"DynamoDB reminder migration verification failed for service '{serviceId}': " + + $"legacy={sourceCount}, v2={targetCount}. Cutover was not performed."); + } + + if (!preserveMigrationState) + { + await WriteMigrationState(new(MigrationStatus.Ready, sourceCount, targetCount)); + } + + LogMigrationVerified(logger, serviceId, sourceCount); + } + + private async Task DeleteOrphan(V2ReminderRecord record) + { + var entry = Resolve(record.Item); + var etag = Clone(record.Item[ETAG_PROPERTY_NAME]); + try + { + await storage.WriteTxAsync( + [ + new() { ConditionCheck = CreateMigrationLeaseFence() }, + new() + { + ConditionCheck = new() + { + TableName = options.TableName, + Key = GetLegacyKey(entry.GrainId, entry.ReminderName), + ConditionExpression = $"attribute_not_exists({REMINDER_ID_PROPERTY_NAME})", + }, + }, + new() + { + Delete = new() + { + TableName = v2TableName, + Key = new() + { + [V2PartitionKeyName] = record.Item[V2PartitionKeyName], + [V2SortKeyName] = record.Item[V2SortKeyName], + }, + ConditionExpression = $"{ETAG_PROPERTY_NAME} = :targetETag", + ExpressionAttributeValues = new() { [":targetETag"] = etag }, + }, + }, + ]); + } + catch (TransactionCanceledException exception) when (IsConditionalFailureAt(exception, 0)) + { + throw new InvalidOperationException("The DynamoDB reminder migration lease was lost while removing an orphan.", exception); + } + catch (TransactionCanceledException exception) when (IsConditionalFailure(exception)) + { + } + } + + private async Task ForEachLegacyRecord(Func action, CancellationToken cancellationToken) + { + Dictionary? checkpoint = null; + do + { + cancellationToken.ThrowIfCancellationRequested(); + var (records, nextCheckpoint) = await storage.ScanPageAsync( + options.TableName, + new() { [":service"] = new(serviceId) }, + $"{SERVICE_ID_PROPERTY_NAME} = :service", + static item => new LegacyReminderRecord(item), + options.MigrationPageSize, + checkpoint, + cancellationToken); + foreach (var record in records) + { + await action(record); + } + + checkpoint = nextCheckpoint; + } + while (checkpoint is { Count: > 0 }); + } + + private async Task ForEachV2Record(Func action, CancellationToken cancellationToken) + { + for (var bucket = 0; bucket < V2BucketCount; bucket++) + { + var values = new Dictionary + { + [":partition"] = new($"{v2DataPartitionPrefix}{bucket:X2}"), + }; + Dictionary? checkpoint = null; + do + { + cancellationToken.ThrowIfCancellationRequested(); + var (records, nextCheckpoint) = await storage.QueryPageAsync( + v2TableName, + values, + $"{V2PartitionKeyName} = :partition", + static item => new V2ReminderRecord(item), + lastEvaluatedKey: checkpoint, + cancellationToken: cancellationToken); + foreach (var record in records) + { + await action(record); + } + + checkpoint = nextCheckpoint; + } + while (checkpoint is { Count: > 0 }); + } + } + + private static bool RecordsEqual( + IReadOnlyDictionary legacy, + IReadOnlyDictionary v2) + => GetAttributeString(legacy[GRAIN_HASH_PROPERTY_NAME]) == GetAttributeString(v2[GRAIN_HASH_PROPERTY_NAME]) + && GetAttributeString(legacy[GRAIN_REFERENCE_PROPERTY_NAME]) == GetAttributeString(v2[GRAIN_REFERENCE_PROPERTY_NAME]) + && GetAttributeString(legacy[REMINDER_NAME_PROPERTY_NAME]) == GetAttributeString(v2[REMINDER_NAME_PROPERTY_NAME]) + && GetAttributeString(legacy[START_TIME_PROPERTY_NAME]) == GetAttributeString(v2[START_TIME_PROPERTY_NAME]) + && GetAttributeString(legacy[PERIOD_PROPERTY_NAME]) == GetAttributeString(v2[PERIOD_PROPERTY_NAME]) + && GetAttributeString(legacy[ETAG_PROPERTY_NAME]) == GetAttributeString(v2[ETAG_PROPERTY_NAME]); + + private async Task AcquireMigrationLease(bool wait, CancellationToken cancellationToken) + { + do + { + if (await TryAcquireMigrationLease()) + { + return true; + } + + if (!wait) + { + return false; + } + + await Task.Delay(TimeSpan.FromSeconds(1), timeProvider, cancellationToken); + } + while (true); + } + + private async Task TryAcquireMigrationLease() + { + var now = timeProvider.GetUtcNow(); + leaseToken ??= Guid.NewGuid().ToString("N"); + var item = MetadataItem(MigrationLeaseSortKey); + item[OwnerAttribute] = new(migrationOwner); + item[LeaseTokenAttribute] = new(leaseToken); + item[ExpiresAtAttribute] = new() { N = now.Add(MigrationLeaseDuration).ToUnixTimeMilliseconds().ToString(CultureInfo.InvariantCulture) }; + try + { + await storage.PutEntryAsync( + v2TableName, + item, + $"attribute_not_exists({OwnerAttribute}) OR {ExpiresAtAttribute} < :now OR ({OwnerAttribute} = :owner AND {LeaseTokenAttribute} = :token)", + new() + { + [":now"] = new() { N = now.ToUnixTimeMilliseconds().ToString(CultureInfo.InvariantCulture) }, + [":owner"] = new(migrationOwner), + [":token"] = new(leaseToken), + }); + return true; + } + catch (ConditionalCheckFailedException) + { + return false; + } + } + + private async Task RenewMigrationLease() + { + if (!await TryAcquireMigrationLease()) + { + throw new InvalidOperationException("The DynamoDB reminder migration lease was lost."); + } + + } + + private void StartLeaseRenewal() + { + leaseRenewalCancellation = new(); + leaseRenewalTask = RunLeaseRenewal(leaseRenewalCancellation.Token); + } + + private async Task RunLeaseRenewal(CancellationToken cancellationToken) + { + try + { + using var timer = new PeriodicTimer(CompatibilityHeartbeatPeriod, timeProvider); + while (await timer.WaitForNextTickAsync(cancellationToken)) + { + await RenewMigrationLease(); + } + } + catch (OperationCanceledException) when (cancellationToken.IsCancellationRequested) + { + } + } + + private async Task StopLeaseRenewal() + { + if (leaseRenewalCancellation is null) + { + return; + } + + await leaseRenewalCancellation.CancelAsync(); + if (leaseRenewalTask is not null) + { + await leaseRenewalTask; + } + + leaseRenewalCancellation.Dispose(); + leaseRenewalCancellation = null; + leaseRenewalTask = null; + } + + private async Task ReleaseMigrationLease() + { + try + { + await storage.DeleteEntryAsync( + v2TableName, + MetadataKey(MigrationLeaseSortKey), + $"{OwnerAttribute} = :owner AND {LeaseTokenAttribute} = :token", + new() + { + [":owner"] = new(migrationOwner), + [":token"] = new(leaseToken), + }); + } + catch (ConditionalCheckFailedException) + { + } + } + + private async Task StartCompatibilityHeartbeat() + { + await WriteCompatibilityMarker(); + heartbeatCancellation = new(); + heartbeatTask = RunCompatibilityHeartbeat(heartbeatCancellation.Token); + } + + private async Task RunCompatibilityHeartbeat(CancellationToken cancellationToken) + { + try + { + using var timer = new PeriodicTimer(CompatibilityHeartbeatPeriod, timeProvider); + while (await timer.WaitForNextTickAsync(cancellationToken)) + { + await WriteCompatibilityMarker(); + } + } + catch (OperationCanceledException) when (cancellationToken.IsCancellationRequested) + { + } + } + + private Task WriteCompatibilityMarker() + { + var item = MetadataItem($"{NodeSortKeyPrefix}{Encode(migrationOwner)}"); + item[OwnerAttribute] = new(migrationOwner); + item[ExpiresAtAttribute] = new() + { + N = timeProvider.GetUtcNow().Add(CompatibilityMarkerLifetime).ToUnixTimeMilliseconds().ToString(CultureInfo.InvariantCulture), + }; + return storage.PutEntryAsync(v2TableName, item); + } + + private async Task EnsureCompatibleCluster(CancellationToken cancellationToken) + { + if (membershipService is null || localSiloDetails is null) + { + throw new InvalidOperationException( + "V2 cutover and rollback require Orleans cluster membership services so active incompatible silos can be detected."); + } + + await membershipService.Refresh(cancellationToken: cancellationToken); + var snapshot = membershipService.CurrentSnapshot; + var markers = await storage.QueryAllAsync( + v2TableName, + new() + { + [":partition"] = new(v2MetadataPartitionKey), + [":prefix"] = new(NodeSortKeyPrefix), + }, + $"{V2PartitionKeyName} = :partition AND begins_with({V2SortKeyName}, :prefix)", + static item => item); + var now = timeProvider.GetUtcNow().ToUnixTimeMilliseconds(); + var compatible = markers + .Where(item => long.Parse(item[ExpiresAtAttribute].N, CultureInfo.InvariantCulture) >= now) + .Select(item => item[OwnerAttribute].S) + .ToHashSet(StringComparer.Ordinal); + var incompatible = snapshot.Members.Values + .Where(static member => member.Status is SiloStatus.Created or SiloStatus.Joining or SiloStatus.Active) + .Select(static member => member.SiloAddress.ToParsableString()) + .Where(address => !compatible.Contains(address)) + .ToArray(); + if (incompatible.Length > 0) + { + throw new InvalidOperationException( + "DynamoDB reminder schema transition was blocked because active silos did not publish V2 compatibility markers: " + + string.Join(", ", incompatible)); + } + + return snapshot.Version; + } + + private async Task ReadMigrationState() + => await storage.ReadSingleEntryAsync(v2TableName, MetadataKey(MigrationStateSortKey), MigrationState.FromItem); + + private async Task WriteMigrationState(MigrationState state) + { + if (leaseToken is null) + { + throw new InvalidOperationException("Migration state cannot be changed without a fencing token."); + } + + try + { + await storage.WriteTxAsync( + [ + new() + { + ConditionCheck = new() + { + TableName = v2TableName, + Key = MetadataKey(MigrationLeaseSortKey), + ConditionExpression = $"{OwnerAttribute} = :owner AND {LeaseTokenAttribute} = :token AND {ExpiresAtAttribute} > :now", + ExpressionAttributeValues = new() + { + [":owner"] = new(migrationOwner), + [":token"] = new(leaseToken), + [":now"] = new() { N = timeProvider.GetUtcNow().ToUnixTimeMilliseconds().ToString(CultureInfo.InvariantCulture) }, + }, + }, + }, + new() + { + Put = new() + { + TableName = v2TableName, + Item = state.ToItem(v2MetadataPartitionKey), + }, + }, + ]); + } + catch (TransactionCanceledException exception) when (IsConditionalFailure(exception)) + { + throw new InvalidOperationException("The DynamoDB reminder migration lease fencing token is no longer current.", exception); + } + } + + private ConditionCheck CreateMigrationLeaseFence() + { + if (leaseToken is null) + { + throw new InvalidOperationException("Migration writes require a fencing token."); + } + + return new() + { + TableName = v2TableName, + Key = MetadataKey(MigrationLeaseSortKey), + ConditionExpression = $"{OwnerAttribute} = :owner AND {LeaseTokenAttribute} = :token AND {ExpiresAtAttribute} > :now", + ExpressionAttributeValues = new() + { + [":owner"] = new(migrationOwner), + [":token"] = new(leaseToken), + [":now"] = new() { N = timeProvider.GetUtcNow().ToUnixTimeMilliseconds().ToString(CultureInfo.InvariantCulture) }, + }, + }; + } + + private Dictionary MetadataItem(string sortKey) + => new() + { + [V2PartitionKeyName] = new(v2MetadataPartitionKey), + [V2SortKeyName] = new(sortKey), + }; + + private Dictionary MetadataKey(string sortKey) => MetadataItem(sortKey); + + private static AttributeValue Clone(AttributeValue value) + => new() { S = value.S, N = value.N, B = value.B }; + + private Dictionary CreateV2ItemFromLegacy( + Dictionary legacy, + ReminderEntry entry) + { + var result = legacy + .Where(static pair => pair.Key != REMINDER_ID_PROPERTY_NAME) + .ToDictionary(static pair => pair.Key, static pair => Clone(pair.Value), StringComparer.Ordinal); + var hash = entry.GrainId.GetUniformHashCode(); + result[V2PartitionKeyName] = new(GetV2PartitionKey(hash)); + result[V2SortKeyName] = new(GetV2SortKey(hash, entry.GrainId, entry.ReminderName)); + return result; + } + + private static string GetAttributeString(AttributeValue value) + => value.S ?? value.N ?? throw new InvalidOperationException("Expected a scalar DynamoDB attribute."); + + private sealed class LegacyReminderRecord(Dictionary item) + { + public Dictionary Item { get; } = item; + + public string Identity => $"{Item[GRAIN_REFERENCE_PROPERTY_NAME].S}\0{Item[REMINDER_NAME_PROPERTY_NAME].S}"; + } + + private sealed class V2ReminderRecord(Dictionary item) + { + public Dictionary Item { get; } = item; + + public string Identity => $"{Item[GRAIN_REFERENCE_PROPERTY_NAME].S}\0{Item[REMINDER_NAME_PROPERTY_NAME].S}"; + } + + internal enum MigrationStatus + { + Backfilling, + Verifying, + Ready, + Cutover, + RolledBack, + VerificationFailed, + Retired, + } + + private sealed class MigrationState(MigrationStatus status, int sourceCount = 0, int targetCount = 0) + { + public MigrationStatus Status { get; set; } = status; + + public string? CheckpointReminderId { get; set; } + + public string? CheckpointGrainHash { get; set; } + + public int SourceCount { get; set; } = sourceCount; + + public int TargetCount { get; set; } = targetCount; + + public Dictionary? GetCheckpoint() + => CheckpointReminderId is null || CheckpointGrainHash is null + ? null + : new() + { + [REMINDER_ID_PROPERTY_NAME] = new(CheckpointReminderId), + [GRAIN_HASH_PROPERTY_NAME] = new() { N = CheckpointGrainHash }, + }; + + public void SetCheckpoint(Dictionary? checkpoint) + { + CheckpointReminderId = checkpoint is { Count: > 0 } ? checkpoint[REMINDER_ID_PROPERTY_NAME].S : null; + CheckpointGrainHash = checkpoint is { Count: > 0 } ? checkpoint[GRAIN_HASH_PROPERTY_NAME].N : null; + } + + public Dictionary ToItem(string partitionKey) + { + var result = new Dictionary + { + [V2PartitionKeyName] = new(partitionKey), + [V2SortKeyName] = new(MigrationStateSortKey), + [StatusAttribute] = new(Status.ToString()), + [SourceCountAttribute] = new() { N = SourceCount.ToString(CultureInfo.InvariantCulture) }, + [TargetCountAttribute] = new() { N = TargetCount.ToString(CultureInfo.InvariantCulture) }, + }; + if (CheckpointReminderId is not null && CheckpointGrainHash is not null) + { + result[CheckpointReminderIdAttribute] = new(CheckpointReminderId); + result[CheckpointGrainHashAttribute] = new() { N = CheckpointGrainHash }; + } + + return result; + } + + public static MigrationState FromItem(Dictionary item) + => new( + Enum.Parse(item[StatusAttribute].S), + int.Parse(item[SourceCountAttribute].N, CultureInfo.InvariantCulture), + int.Parse(item[TargetCountAttribute].N, CultureInfo.InvariantCulture)) + { + CheckpointReminderId = item.TryGetValue(CheckpointReminderIdAttribute, out var reminderId) ? reminderId.S : null, + CheckpointGrainHash = item.TryGetValue(CheckpointGrainHashAttribute, out var grainHash) ? grainHash.N : null, + }; + } + + [LoggerMessage(Level = LogLevel.Information, Message = "DynamoDB reminder migration for service {ServiceId} entered {State}.")] + private static partial void LogMigrationState(ILogger logger, string serviceId, string state); + + [LoggerMessage(Level = LogLevel.Information, Message = "DynamoDB reminder migration for service {ServiceId} completed page {PageNumber} containing {ItemCount} matching items.")] + private static partial void LogMigrationPage(ILogger logger, string serviceId, int pageNumber, int itemCount); + + [LoggerMessage(Level = LogLevel.Information, Message = "DynamoDB reminder migration for service {ServiceId} verified {ItemCount} reminders.")] + private static partial void LogMigrationVerified(ILogger logger, string serviceId, int itemCount); + + [LoggerMessage(Level = LogLevel.Error, Message = "DynamoDB reminder migration verification failed for service {ServiceId}: source count {SourceCount}, target count {TargetCount}.")] + private static partial void LogMigrationVerificationFailed(ILogger logger, string serviceId, int sourceCount, int targetCount); + + [LoggerMessage(Level = LogLevel.Information, Message = "DynamoDB reminder migration lease for service {ServiceId} is held by another owner; {Owner} will continue in the current read mode.")] + private static partial void LogMigrationLeaseContended(ILogger logger, string serviceId, string owner); +} diff --git a/src/AWS/Orleans.Reminders.DynamoDB/Reminders/DynamoDBReminderTable.Schema.cs b/src/AWS/Orleans.Reminders.DynamoDB/Reminders/DynamoDBReminderTable.Schema.cs new file mode 100644 index 00000000000..b82ce892b98 --- /dev/null +++ b/src/AWS/Orleans.Reminders.DynamoDB/Reminders/DynamoDBReminderTable.Schema.cs @@ -0,0 +1,366 @@ +using Amazon.DynamoDBv2; +using Amazon.DynamoDBv2.Model; +using Orleans.Runtime; +using System.Globalization; +using System.Security.Cryptography; +using System.Text; + +namespace Orleans.Reminders.DynamoDB; + +internal sealed partial class DynamoDBReminderTable +{ + internal const int V2BucketCount = 32; + internal const string V2PartitionKeyName = "PartitionKey"; + internal const string V2SortKeyName = "SortKey"; + + private const string DataPartitionPrefix = "D#"; + private const string MetadataPartitionPrefix = "M#"; + private const string ReminderSortPrefix = "R#"; + + private async Task InitializeV2Table(CancellationToken cancellationToken) + { + await storage.InitializeTable( + v2TableName, + [ + new() { AttributeName = V2PartitionKeyName, KeyType = KeyType.HASH }, + new() { AttributeName = V2SortKeyName, KeyType = KeyType.RANGE }, + ], + [ + new() { AttributeName = V2PartitionKeyName, AttributeType = ScalarAttributeType.S }, + new() { AttributeName = V2SortKeyName, AttributeType = ScalarAttributeType.S }, + ], + cancellationToken: cancellationToken); + } + + internal static string GetV2PartitionKey(string serviceId, uint grainHash) + => $"{DataPartitionPrefix}{Encode(serviceId)}#{grainHash % V2BucketCount:X2}"; + + private string GetV2PartitionKey(uint grainHash) => $"{v2DataPartitionPrefix}{grainHash % V2BucketCount:X2}"; + + internal static string GetV2SortKey(uint grainHash, GrainId grainId, string reminderName) + => $"{GetV2GrainPrefix(grainHash, grainId)}{EncodeKeyComponent(reminderName, 600)}"; + + internal static string GetV2GrainPrefix(uint grainHash, GrainId grainId) + => $"{ReminderSortPrefix}{grainHash:X8}#{EncodeKeyComponent(grainId.ToString(), 300)}#"; + + internal static (string Lower, string Upper) GetV2RangeBounds(uint lowerInclusive, uint upperInclusive) + => ($"{ReminderSortPrefix}{lowerInclusive:X8}#", $"{ReminderSortPrefix}{upperInclusive:X8}#~"); + + private static string Encode(string value) => EncodeKeyComponent(value, 300); + + private static string EncodeKeyComponent(string value, int maximumEncodedLength) + { + var bytes = Encoding.UTF8.GetBytes(value); + var encoded = Base64Url(bytes); + return encoded.Length <= maximumEncodedLength + ? $"V{encoded}" + : $"H{Base64Url(SHA256.HashData(bytes))}"; + } + + private static string Base64Url(ReadOnlySpan value) + => Convert.ToBase64String(value).TrimEnd('=').Replace('+', '-').Replace('/', '_'); + + private Dictionary GetLegacyKey(GrainId grainId, string reminderName) + => new() + { + [REMINDER_ID_PROPERTY_NAME] = new(ConstructReminderId(serviceId, grainId, reminderName)), + [GRAIN_HASH_PROPERTY_NAME] = new() { N = grainId.GetUniformHashCode().ToString(CultureInfo.InvariantCulture) }, + }; + + private Dictionary GetV2Key(GrainId grainId, string reminderName) + { + var hash = grainId.GetUniformHashCode(); + return new() + { + [V2PartitionKeyName] = new(GetV2PartitionKey(hash)), + [V2SortKeyName] = new(GetV2SortKey(hash, grainId, reminderName)), + }; + } + + private Dictionary CreateLegacyItem(ReminderEntry entry, string etag) + => new() + { + [REMINDER_ID_PROPERTY_NAME] = new(ConstructReminderId(serviceId, entry.GrainId, entry.ReminderName)), + [GRAIN_HASH_PROPERTY_NAME] = new() { N = entry.GrainId.GetUniformHashCode().ToString(CultureInfo.InvariantCulture) }, + [SERVICE_ID_PROPERTY_NAME] = new(serviceId), + [GRAIN_REFERENCE_PROPERTY_NAME] = new(entry.GrainId.ToString()), + [PERIOD_PROPERTY_NAME] = new(entry.Period.ToString()), + [START_TIME_PROPERTY_NAME] = new(entry.StartAt.ToString("O", CultureInfo.InvariantCulture)), + [REMINDER_NAME_PROPERTY_NAME] = new(entry.ReminderName), + [ETAG_PROPERTY_NAME] = new() { N = etag }, + }; + + private Dictionary CreateV2Item(ReminderEntry entry, string etag) + { + var item = CreateLegacyItem(entry, etag); + item.Remove(REMINDER_ID_PROPERTY_NAME); + var hash = entry.GrainId.GetUniformHashCode(); + item[V2PartitionKeyName] = new(GetV2PartitionKey(hash)); + item[V2SortKeyName] = new(GetV2SortKey(hash, entry.GrainId, entry.ReminderName)); + return item; + } + + private async Task ReadV2Row(GrainId grainId, string reminderName) + => await storage.ReadSingleEntryAsync(v2TableName, GetV2Key(grainId, reminderName), Resolve); + + private async Task ReadV2Rows(GrainId grainId) + { + var hash = grainId.GetUniformHashCode(); + var values = new Dictionary + { + [":partition"] = new(GetV2PartitionKey(hash)), + [":prefix"] = new(GetV2GrainPrefix(hash, grainId)), + }; + var rows = await storage.QueryAllAsync( + v2TableName, + values, + $"{V2PartitionKeyName} = :partition AND begins_with({V2SortKeyName}, :prefix)", + Resolve); + return new(rows); + } + + private async Task ReadV2Rows(uint begin, uint end) + { + var tasks = new List>>(); + for (var bucket = 0; bucket < V2BucketCount; bucket++) + { + if (begin < end) + { + tasks.Add(QueryV2RangeBucket(bucket, begin + 1, end)); + } + else + { + if (begin != uint.MaxValue) + { + tasks.Add(QueryV2RangeBucket(bucket, begin + 1, uint.MaxValue)); + } + + tasks.Add(QueryV2RangeBucket(bucket, 0, end)); + } + } + + var pages = await Task.WhenAll(tasks); + return new(pages.SelectMany(static page => page)); + } + + private Task> QueryV2RangeBucket(int bucket, uint lowerInclusive, uint upperInclusive) + { + var bounds = GetV2RangeBounds(lowerInclusive, upperInclusive); + var values = new Dictionary + { + [":partition"] = new($"{v2DataPartitionPrefix}{bucket:X2}"), + [":lower"] = new(bounds.Lower), + [":upper"] = new(bounds.Upper), + }; + return storage.QueryAllAsync( + v2TableName, + values, + $"{V2PartitionKeyName} = :partition AND {V2SortKeyName} BETWEEN :lower AND :upper", + Resolve); + } + + private async Task UpsertDualRow(ReminderEntry entry) + { + var etag = Random.Shared.NextInt64(1, long.MaxValue).ToString(CultureInfo.InvariantCulture); + var legacy = CreateLegacyItem(entry, etag); + var v2 = CreateV2Item(entry, etag); + try + { + await storage.WriteTxAsync( + [ + new() { ConditionCheck = CreateDualWriteFence() }, + new() + { + Put = new() { TableName = options.TableName, Item = legacy }, + }, + new() + { + Put = CreateIdentitySafeV2Put(v2, entry), + }, + ]); + } + catch (TransactionCanceledException exception) when (IsConditionalFailureAt(exception, 0)) + { + await RefreshReadMode(); + if (useV2OnlyWrites) + { + return await UpsertV2OnlyRow(entry); + } + + throw; + } + catch (TransactionCanceledException exception) when (IsTransactionConflict(exception)) + { + await RefreshReadMode(); + if (useV2OnlyWrites) + { + return await UpsertV2OnlyRow(entry); + } + + throw; + } + + entry.ETag = etag; + return etag; + } + + private async Task UpsertV2OnlyRow(ReminderEntry entry) + { + var etag = Random.Shared.NextInt64(1, long.MaxValue).ToString(CultureInfo.InvariantCulture); + await storage.WriteTxAsync( + [ + new() + { + Put = CreateIdentitySafeV2Put(CreateV2Item(entry, etag), entry), + }, + ]); + entry.ETag = etag; + return etag; + } + + private Put CreateIdentitySafeV2Put(Dictionary item, ReminderEntry entry) + => new() + { + TableName = v2TableName, + Item = item, + ConditionExpression = $"attribute_not_exists({V2PartitionKeyName}) OR ({SERVICE_ID_PROPERTY_NAME} = :service AND {GRAIN_REFERENCE_PROPERTY_NAME} = :grain AND {REMINDER_NAME_PROPERTY_NAME} = :reminder)", + ExpressionAttributeValues = new() + { + [":service"] = new(serviceId), + [":grain"] = new(entry.GrainId.ToString()), + [":reminder"] = new(entry.ReminderName), + }, + }; + + private async Task RemoveDualRow(GrainId grainId, string reminderName, string etag) + { + var values = new Dictionary { [CURRENT_ETAG_ALIAS] = new() { N = etag } }; + try + { + await storage.WriteTxAsync( + [ + new() { ConditionCheck = CreateDualWriteFence() }, + new() + { + Delete = new() + { + TableName = options.TableName, + Key = GetLegacyKey(grainId, reminderName), + ConditionExpression = $"{ETAG_PROPERTY_NAME} = {CURRENT_ETAG_ALIAS}", + ExpressionAttributeValues = values, + }, + }, + new() + { + Delete = new() + { + TableName = v2TableName, + Key = GetV2Key(grainId, reminderName), + }, + }, + ]); + return true; + } + catch (TransactionCanceledException exception) when (IsConditionalFailureAt(exception, 0)) + { + await RefreshReadMode(); + if (useV2OnlyWrites) + { + return await RemoveV2OnlyRow(grainId, reminderName, etag); + } + + throw; + } + catch (TransactionCanceledException exception) when (IsTransactionConflict(exception)) + { + await RefreshReadMode(); + if (useV2OnlyWrites) + { + return await RemoveV2OnlyRow(grainId, reminderName, etag); + } + + throw; + } + catch (TransactionCanceledException exception) when (IsConditionalFailure(exception)) + { + return false; + } + } + + private async Task RemoveV2OnlyRow(GrainId grainId, string reminderName, string etag) + { + try + { + await storage.WriteTxAsync( + [ + new() + { + Delete = new() + { + TableName = v2TableName, + Key = GetV2Key(grainId, reminderName), + ConditionExpression = $"{ETAG_PROPERTY_NAME} = {CURRENT_ETAG_ALIAS}", + ExpressionAttributeValues = new() + { + [CURRENT_ETAG_ALIAS] = new() { N = etag }, + }, + }, + }, + ]); + return true; + } + catch (TransactionCanceledException exception) when (IsConditionalFailure(exception)) + { + return false; + } + } + + private ConditionCheck CreateDualWriteFence() + => new() + { + TableName = v2TableName, + Key = MetadataKey(MigrationStateSortKey), + ConditionExpression = $"attribute_not_exists({StatusAttribute}) OR {StatusAttribute} <> :retired", + ExpressionAttributeValues = new() + { + [":retired"] = new(MigrationStatus.Retired.ToString()), + }, + }; + + private async Task ClearV2ServiceRows() + { + var keys = new List>(); + for (var bucket = 0; bucket < V2BucketCount; bucket++) + { + var values = new Dictionary + { + [":partition"] = new($"{v2DataPartitionPrefix}{bucket:X2}"), + }; + keys.AddRange(await storage.QueryAllAsync( + v2TableName, + values, + $"{V2PartitionKeyName} = :partition", + static item => new Dictionary + { + [V2PartitionKeyName] = item[V2PartitionKeyName], + [V2SortKeyName] = item[V2SortKeyName], + })); + } + + foreach (var batch in keys.BatchIEnumerable(25)) + { + await storage.DeleteEntriesAsync(v2TableName, batch); + } + } + + private static bool IsConditionalFailure(TransactionCanceledException exception) + => exception.CancellationReasons?.Any(static reason => reason.Code == "ConditionalCheckFailed") == true; + + private static bool IsConditionalFailureAt(TransactionCanceledException exception, int index) + => exception.CancellationReasons is { } reasons + && reasons.Count > index + && reasons[index].Code == "ConditionalCheckFailed"; + + private static bool IsTransactionConflict(TransactionCanceledException exception) + => exception.CancellationReasons?.Any(static reason => reason.Code == "TransactionConflict") == true; +} diff --git a/src/AWS/Orleans.Reminders.DynamoDB/Reminders/DynamoDBReminderTable.cs b/src/AWS/Orleans.Reminders.DynamoDB/Reminders/DynamoDBReminderTable.cs index fb40964dbb7..bd8ea1b6b11 100644 --- a/src/AWS/Orleans.Reminders.DynamoDB/Reminders/DynamoDBReminderTable.cs +++ b/src/AWS/Orleans.Reminders.DynamoDB/Reminders/DynamoDBReminderTable.cs @@ -6,6 +6,7 @@ using Orleans.Runtime; using System; using System.Collections.Generic; +using System.Threading; using System.Threading.Tasks; namespace Orleans.Reminders.DynamoDB @@ -30,8 +31,23 @@ internal sealed partial class DynamoDBReminderTable : IReminderTable private readonly ILogger logger; private readonly DynamoDBReminderStorageOptions options; private readonly string serviceId; + private readonly IClusterMembershipService? membershipService; + private readonly ILocalSiloDetails? localSiloDetails; + private readonly TimeProvider timeProvider; + private readonly DynamoDBReminderMigrationTestHooks? testHooks; + private readonly string migrationOwner; + private readonly string v2TableName; + private readonly string v2DataPartitionPrefix; + private readonly string v2MetadataPartitionKey; private DynamoDBStorage storage = null!; + private bool useV2Reads; + private bool useV2OnlyWrites; + private CancellationTokenSource? heartbeatCancellation; + private Task? heartbeatTask; + private string? leaseToken; + private CancellationTokenSource? leaseRenewalCancellation; + private Task? leaseRenewalTask; /// Initializes a new instance of the class. /// logger factory to use @@ -40,16 +56,43 @@ internal sealed partial class DynamoDBReminderTable : IReminderTable public DynamoDBReminderTable( ILoggerFactory loggerFactory, IOptions clusterOptions, - IOptions storageOptions) + IOptions storageOptions, + IClusterMembershipService? membershipService = null, + ILocalSiloDetails? localSiloDetails = null, + TimeProvider? timeProvider = null) + : this(loggerFactory, clusterOptions, storageOptions, membershipService, localSiloDetails, timeProvider, null) + { + } + + internal DynamoDBReminderTable( + ILoggerFactory loggerFactory, + IOptions clusterOptions, + IOptions storageOptions, + IClusterMembershipService? membershipService, + ILocalSiloDetails? localSiloDetails, + TimeProvider? timeProvider, + DynamoDBReminderMigrationTestHooks? testHooks) { this.logger = loggerFactory.CreateLogger(); this.serviceId = clusterOptions.Value.ServiceId; this.options = storageOptions.Value; + this.membershipService = membershipService; + this.localSiloDetails = localSiloDetails; + this.timeProvider = timeProvider ?? TimeProvider.System; + this.testHooks = testHooks; + this.migrationOwner = localSiloDetails?.SiloAddress.ToParsableString() ?? Guid.NewGuid().ToString("N"); + this.v2TableName = options.V2TableName ?? $"{options.TableName}-v2"; + this.v2DataPartitionPrefix = $"{DataPartitionPrefix}{Encode(serviceId)}#"; + this.v2MetadataPartitionKey = $"{MetadataPartitionPrefix}{Encode(serviceId)}"; } /// Initialize current instance with specific global configuration and logger - public Task Init() + public Task Init() => StartAsync(); + + /// + public async Task StartAsync(CancellationToken cancellationToken = default) { + ValidateOptions(); this.storage = new DynamoDBStorage( this.logger, this.options.Service, @@ -65,6 +108,25 @@ public Task Init() LogInformationInitializingDynamoDBRemindersTable(logger); + if (options.TableMode != DynamoDBReminderTableMode.Legacy) + { + await InitializeV2Table(cancellationToken); + if ((await ReadMigrationState())?.Status == MigrationStatus.Retired) + { + await StartCompatibilityHeartbeat(); + try + { + await InitializeMigration(cancellationToken); + return; + } + catch + { + await StopAsync(); + throw; + } + } + } + var serviceIdGrainHashGlobalSecondaryIndex = new GlobalSecondaryIndex { IndexName = SERVICE_ID_GRAIN_HASH_INDEX, @@ -87,7 +149,7 @@ public Task Init() } }; - return this.storage.InitializeTable(this.options.TableName, + await this.storage.InitializeTable(this.options.TableName, new List { new KeySchemaElement { AttributeName = REMINDER_ID_PROPERTY_NAME, KeyType = KeyType.HASH }, @@ -100,7 +162,25 @@ public Task Init() new AttributeDefinition { AttributeName = SERVICE_ID_PROPERTY_NAME, AttributeType = ScalarAttributeType.S }, new AttributeDefinition { AttributeName = GRAIN_REFERENCE_PROPERTY_NAME, AttributeType = ScalarAttributeType.S } }, - new List { serviceIdGrainHashGlobalSecondaryIndex, serviceIdGrainReferenceGlobalSecondaryIndex }); + new List { serviceIdGrainHashGlobalSecondaryIndex, serviceIdGrainReferenceGlobalSecondaryIndex }, + cancellationToken: cancellationToken); + + if (options.TableMode == DynamoDBReminderTableMode.Legacy) + { + return; + } + + await StartCompatibilityHeartbeat(); + + try + { + await InitializeMigration(cancellationToken); + } + catch + { + await StopAsync(); + throw; + } } /// @@ -112,6 +192,12 @@ public Task Init() /// Return the ReminderTableData if the rows were read successfully public async Task ReadRow(GrainId grainId, string reminderName) { + await RefreshReadMode(); + if (useV2Reads) + { + return await ReadV2Row(grainId, reminderName); + } + var reminderId = ConstructReminderId(this.serviceId, grainId, reminderName); var keys = new Dictionary @@ -138,6 +224,12 @@ public Task Init() /// Return the ReminderTableData if the rows were read successfully public async Task ReadRows(GrainId grainId) { + await RefreshReadMode(); + if (useV2Reads) + { + return await ReadV2Rows(grainId); + } + var expressionValues = new Dictionary { { $":{SERVICE_ID_PROPERTY_NAME}", new AttributeValue(this.serviceId) }, @@ -148,6 +240,7 @@ public async Task ReadRows(GrainId grainId) { var expression = $"{SERVICE_ID_PROPERTY_NAME} = :{SERVICE_ID_PROPERTY_NAME} AND {GRAIN_REFERENCE_PROPERTY_NAME} = :{GRAIN_REFERENCE_PROPERTY_NAME}"; var records = await this.storage.QueryAllAsync(this.options.TableName, expressionValues, expression, this.Resolve, SERVICE_ID_GRAIN_REFERENCE_INDEX, consistentRead: false).ConfigureAwait(false); + records = await ConfirmLegacyDiscoveryCandidates(records); return new ReminderTableData(records); } @@ -166,6 +259,12 @@ public async Task ReadRows(GrainId grainId) /// Return the RemiderTableData if the rows were read successfully public async Task ReadRows(uint begin, uint end) { + await RefreshReadMode(); + if (useV2Reads) + { + return await ReadV2Rows(begin, end); + } + Dictionary? expressionValues = null; try @@ -204,6 +303,7 @@ public async Task ReadRows(uint begin, uint end) } + records = await ConfirmLegacyDiscoveryCandidates(records); return new ReminderTableData(records); } catch (Exception exc) @@ -217,7 +317,7 @@ private ReminderEntry Resolve(Dictionary item) { return new ReminderEntry { - ETag = item[ETAG_PROPERTY_NAME].N, + ETag = item[ETAG_PROPERTY_NAME].S ?? item[ETAG_PROPERTY_NAME].N, GrainId = GrainId.Parse(item[GRAIN_REFERENCE_PROPERTY_NAME].S), Period = TimeSpan.Parse(item[PERIOD_PROPERTY_NAME].S), ReminderName = item[REMINDER_NAME_PROPERTY_NAME].S, @@ -234,6 +334,17 @@ private ReminderEntry Resolve(Dictionary item) /// Return true if the row was removed public async Task RemoveRow(GrainId grainId, string reminderName, string eTag) { + await RefreshReadMode(); + if (useV2OnlyWrites) + { + return await RemoveV2OnlyRow(grainId, reminderName, eTag); + } + + if (options.TableMode != DynamoDBReminderTableMode.Legacy) + { + return await RemoveDualRow(grainId, reminderName, eTag); + } + var reminderId = ConstructReminderId(this.serviceId, grainId, reminderName); var keys = new Dictionary @@ -262,6 +373,17 @@ public async Task RemoveRow(GrainId grainId, string reminderName, string e /// public async Task TestOnlyClearTable() { + await RefreshReadMode(); + if (options.TableMode != DynamoDBReminderTableMode.Legacy) + { + await ClearV2ServiceRows(); + } + + if (useV2OnlyWrites) + { + return; + } + var expressionValues = new Dictionary { { $":{SERVICE_ID_PROPERTY_NAME}", new AttributeValue(this.serviceId) } @@ -305,6 +427,17 @@ public async Task TestOnlyClearTable() /// Return the entry ETag if entry was upsert successfully public async Task UpsertRow(ReminderEntry entry) { + await RefreshReadMode(); + if (useV2OnlyWrites) + { + return await UpsertV2OnlyRow(entry); + } + + if (options.TableMode != DynamoDBReminderTableMode.Legacy) + { + return await UpsertDualRow(entry); + } + var reminderId = ConstructReminderId(this.serviceId, entry.GrainId, entry.ReminderName); var fields = new Dictionary @@ -337,6 +470,25 @@ public async Task TestOnlyClearTable() private static string ConstructReminderId(string serviceId, GrainId grainId, string reminderName) => $"{serviceId}_{grainId}_{reminderName}"; + /// + public async Task StopAsync(CancellationToken cancellationToken = default) + { + if (heartbeatCancellation is null) + { + return; + } + + await heartbeatCancellation.CancelAsync(); + if (heartbeatTask is not null) + { + await heartbeatTask.WaitAsync(cancellationToken); + } + + heartbeatCancellation.Dispose(); + heartbeatCancellation = null; + heartbeatTask = null; + } + [LoggerMessage( EventId = (int)ErrorCode.ReminderServiceBase, Level = LogLevel.Information, diff --git a/src/AWS/Orleans.Reminders.DynamoDB/Reminders/DynamoDbReminderStorageOptions.cs b/src/AWS/Orleans.Reminders.DynamoDB/Reminders/DynamoDbReminderStorageOptions.cs index e43d8a902a1..05ce7d790de 100644 --- a/src/AWS/Orleans.Reminders.DynamoDB/Reminders/DynamoDbReminderStorageOptions.cs +++ b/src/AWS/Orleans.Reminders.DynamoDB/Reminders/DynamoDbReminderStorageOptions.cs @@ -2,6 +2,38 @@ namespace Orleans.Configuration { + /// + /// Controls the DynamoDB reminder schema migration protocol. + /// + public enum DynamoDBReminderTableMode + { + /// + /// Uses the legacy table exclusively. This mode is compatible with older Orleans binaries. + /// + Legacy, + + /// + /// Creates and backfills the V2 table, then writes both schemas atomically while continuing to read V1. + /// + Migrate, + + /// + /// Completes and verifies migration, requires all active silos to be V2-capable, and reads V2. + /// Writes continue to both schemas to retain the rollback window. + /// + V2, + + /// + /// Verifies both schemas and returns reads to V1. Writes continue to both schemas. + /// + Rollback, + + /// + /// Irreversibly retires V1 after verifying it for the last time, then reads and writes only V2. + /// + V2Only, + } + /// /// Configuration for Amazon DynamoDB reminder storage. /// @@ -37,5 +69,20 @@ public class DynamoDBReminderStorageOptions : DynamoDBClientOptions /// Defaults to 'OrleansReminders'. /// public string TableName { get; set; } = "OrleansReminders"; + + /// + /// Gets or sets the V2 table name. When unset, -v2 is appended to . + /// + public string? V2TableName { get; set; } + + /// + /// Gets or sets the schema migration mode. The default is . + /// + public DynamoDBReminderTableMode TableMode { get; set; } + + /// + /// Gets or sets the maximum number of legacy items evaluated by each resumable backfill scan page. + /// + public int MigrationPageSize { get; set; } = 100; } } \ No newline at end of file diff --git a/src/AWS/Orleans.Reminders.DynamoDB/Reminders/DynamoDbReminderStorageOptionsExtensions.cs b/src/AWS/Orleans.Reminders.DynamoDB/Reminders/DynamoDbReminderStorageOptionsExtensions.cs index 3f890f607b9..36b08e8346b 100644 --- a/src/AWS/Orleans.Reminders.DynamoDB/Reminders/DynamoDbReminderStorageOptionsExtensions.cs +++ b/src/AWS/Orleans.Reminders.DynamoDB/Reminders/DynamoDbReminderStorageOptionsExtensions.cs @@ -16,6 +16,10 @@ public static class DynamoDBReminderStorageOptionsExtensions private const string UseProvisionedThroughputPropertyName = "UseProvisionedThroughput"; private const string CreateIfNotExistsPropertyName = "CreateIfNotExists"; private const string UpdateIfExistsPropertyName = "UpdateIfExists"; + private const string TableNamePropertyName = "TableName"; + private const string V2TableNamePropertyName = "V2TableName"; + private const string TableModePropertyName = "TableMode"; + private const string MigrationPageSizePropertyName = "MigrationPageSize"; /// /// Configures this instance using the provided connection string. @@ -87,6 +91,41 @@ public static void ParseConnectionString(this DynamoDBReminderStorageOptions opt if (value.Length == 2 && !string.IsNullOrWhiteSpace(value[1])) options.UpdateIfExists = bool.Parse(value[1]); } + + var tableNameConfig = FindParameter(parameters, TableNamePropertyName); + if (!string.IsNullOrWhiteSpace(tableNameConfig)) + { + var value = tableNameConfig.Split(new[] { '=' }, StringSplitOptions.RemoveEmptyEntries); + if (value.Length == 2 && !string.IsNullOrWhiteSpace(value[1])) + options.TableName = value[1]; + } + + var v2TableNameConfig = FindParameter(parameters, V2TableNamePropertyName); + if (!string.IsNullOrWhiteSpace(v2TableNameConfig)) + { + var value = v2TableNameConfig.Split(new[] { '=' }, StringSplitOptions.RemoveEmptyEntries); + if (value.Length == 2 && !string.IsNullOrWhiteSpace(value[1])) + options.V2TableName = value[1]; + } + + var tableModeConfig = FindParameter(parameters, TableModePropertyName); + if (!string.IsNullOrWhiteSpace(tableModeConfig)) + { + var value = tableModeConfig.Split(new[] { '=' }, StringSplitOptions.RemoveEmptyEntries); + if (value.Length == 2 && !string.IsNullOrWhiteSpace(value[1])) + options.TableMode = Enum.Parse(value[1], ignoreCase: true); + } + + var migrationPageSizeConfig = FindParameter(parameters, MigrationPageSizePropertyName); + if (!string.IsNullOrWhiteSpace(migrationPageSizeConfig)) + { + var value = migrationPageSizeConfig.Split(new[] { '=' }, StringSplitOptions.RemoveEmptyEntries); + if (value.Length == 2 && !string.IsNullOrWhiteSpace(value[1])) + options.MigrationPageSize = int.Parse(value[1]); + } } + + private static string? FindParameter(string[] parameters, string name) + => Array.Find(parameters, parameter => parameter.StartsWith($"{name}=", StringComparison.OrdinalIgnoreCase)); } } \ No newline at end of file diff --git a/src/AWS/Shared/Storage/DynamoDBStorage.cs b/src/AWS/Shared/Storage/DynamoDBStorage.cs index c76c5b78149..6ecac3c87a8 100755 --- a/src/AWS/Shared/Storage/DynamoDBStorage.cs +++ b/src/AWS/Shared/Storage/DynamoDBStorage.cs @@ -723,7 +723,13 @@ public Task DeleteEntriesAsync(string tableName, IReadOnlyCollectionThe primary key of the first item that this operation will evaluate. Use the value that was returned for LastEvaluatedKey in the previous operation /// Determines the read consistency model. Note that if a GSI is used, this must be false. /// The collection containing a list of objects translated by the resolver function and the LastEvaluatedKey for paged results - public async Task<(List results, Dictionary? lastEvaluatedKey)> QueryAsync(string tableName, Dictionary keys, string keyConditionExpression, Func, TResult> resolver, string indexName = "", bool scanIndexForward = true, Dictionary? lastEvaluatedKey = null, bool consistentRead = true) where TResult : class + public Task<(List results, Dictionary? lastEvaluatedKey)> QueryAsync(string tableName, Dictionary keys, string keyConditionExpression, Func, TResult> resolver, string indexName = "", bool scanIndexForward = true, Dictionary? lastEvaluatedKey = null, bool consistentRead = true) where TResult : class + => QueryPageAsync(tableName, keys, keyConditionExpression, resolver, indexName, scanIndexForward, lastEvaluatedKey, consistentRead, CancellationToken.None); + + /// + /// Queries one page of entries with cancellation support. + /// + public async Task<(List results, Dictionary? lastEvaluatedKey)> QueryPageAsync(string tableName, Dictionary keys, string keyConditionExpression, Func, TResult> resolver, string indexName = "", bool scanIndexForward = true, Dictionary? lastEvaluatedKey = null, bool consistentRead = true, CancellationToken cancellationToken = default) where TResult : class { try { @@ -747,7 +753,7 @@ public Task DeleteEntriesAsync(string tableName, IReadOnlyCollection(); foreach (var item in response.Items) @@ -867,6 +873,47 @@ public async Task> ScanAsync(string tableName, Dictionary } } + /// + /// Strongly consistently scans one bounded page from a DynamoDB table. + /// + public async Task<(List results, Dictionary? lastEvaluatedKey)> ScanPageAsync( + string tableName, + Dictionary attributes, + string expression, + Func, TResult> resolver, + int limit, + Dictionary? exclusiveStartKey = null, + CancellationToken cancellationToken = default) + where TResult : class + { + try + { + var request = new ScanRequest + { + TableName = tableName, + ConsistentRead = true, + FilterExpression = expression, + ExpressionAttributeValues = attributes, + Select = Select.ALL_ATTRIBUTES, + Limit = limit, + ExclusiveStartKey = exclusiveStartKey, + }; + + var response = await _ddbClient.ScanAsync(request, cancellationToken); + var results = response.Items?.Select(resolver).ToList() ?? []; + return (results, response.LastEvaluatedKey); + } + catch (OperationCanceledException) + { + throw; + } + catch (Exception exc) + { + LogWarningFailedToReadTable(_logger, exc, tableName); + throw new OrleansException($"Failed to read table {tableName}: {exc.Message}", exc); + } + } + /// /// Crete or replace multiple entries in a DynamoDB table (Batch put) /// diff --git a/src/Orleans.Reminders/OrleansContracts.txt b/src/Orleans.Reminders/OrleansContracts.txt index a948172d16a..f2e53c188c8 100644 --- a/src/Orleans.Reminders/OrleansContracts.txt +++ b/src/Orleans.Reminders/OrleansContracts.txt @@ -16,6 +16,7 @@ interface [GrainInterfaceType("Orleans.IRemindable")] Orleans.IRemindable [Versi interface [GrainInterfaceType("Orleans.IReminderService")] Orleans.IReminderService [Version(0)] AC622EEB: GetReminder(Orleans.Runtime.GrainId, string) -> Task 419EB51E: GetReminders(Orleans.Runtime.GrainId) -> Task> + 902EB550: ReconcileReminder(Orleans.Runtime.GrainId, string, int) -> Task 1281C86D: RegisterOrUpdateReminder(Orleans.Runtime.GrainId, string, System.TimeSpan, System.TimeSpan) -> Task 5CF78F8A: Start() -> Task DCFCA00D: Stop() -> Task diff --git a/src/Orleans.Reminders/ReminderService/LocalReminderService.cs b/src/Orleans.Reminders/ReminderService/LocalReminderService.cs index d66c96592ae..72199939877 100644 --- a/src/Orleans.Reminders/ReminderService/LocalReminderService.cs +++ b/src/Orleans.Reminders/ReminderService/LocalReminderService.cs @@ -17,12 +17,15 @@ namespace Orleans.Runtime.ReminderService internal sealed partial class LocalReminderService : GrainService, IReminderService, ILifecycleParticipant { private const int InitialReadRetryCountBeforeFastFailForUpdates = 2; + private const int MissingReminderPointReadConcurrency = 16; + private const int MutationNotificationMaxHops = 3; private static readonly TimeSpan InitialReadMaxWaitTimeForUpdates = TimeSpan.FromSeconds(20); private static readonly TimeSpan InitialReadRetryPeriod = TimeSpan.FromSeconds(30); private static readonly TimeSpan MinimumReminderDueTime = TimeSpan.FromMilliseconds(1); private readonly ILogger logger; private readonly ReminderOptions reminderOptions; private readonly Dictionary localReminders = new(); + private readonly Dictionary mutationReconciliations = new(); private readonly IReminderTable reminderTable; private readonly TaskCompletionSource startedTask; private readonly IAsyncTimer listRefreshTimer; // timer that refreshes our list of reminders to reflect global reminder table @@ -30,6 +33,9 @@ internal sealed partial class LocalReminderService : GrainService, IReminderServ private readonly GrainInterfaceType _grainInterfaceType; private readonly TimeProvider _timeProvider; private readonly ReminderInstruments _reminderInstruments; + private readonly IConsistentRingProvider _ringProvider; + private readonly IInternalGrainFactory _grainFactory; + private readonly GrainType _reminderServiceGrainType; private long localTableSequence; // The test barrier reads this state off-scheduler so it remains observable while the service is busy. private readonly object _rangeChangeLock = new(); @@ -50,6 +56,7 @@ public LocalReminderService( IAsyncTimerFactory asyncTimerFactory, IOptions reminderOptions, IConsistentRingProvider ringProvider, + IInternalGrainFactory grainFactory, [FromKeyedServices(ReminderTimeProviderNames.Reminders)] TimeProvider timeProvider, ReminderInstruments reminderInstruments, SystemTargetShared shared) @@ -64,6 +71,11 @@ public LocalReminderService( this.reminderTable = reminderTable; _timeProvider = timeProvider; _reminderInstruments = reminderInstruments; + _ringProvider = ringProvider; + _grainFactory = grainFactory; + _reminderServiceGrainType = SystemTargetGrainId.CreateGrainServiceGrainType( + GrainInterfaceUtils.GetGrainClassTypeCode(typeof(IReminderService)), + null); _reminderInstruments.RegisterActiveRemindersObserve(() => localReminders.Count); startedTask = new TaskCompletionSource(TaskCreationOptions.RunContinuationsAsynchronously); this.logger = shared.LoggerFactory.CreateLogger(); @@ -222,11 +234,12 @@ public async Task RegisterOrUpdateReminder(GrainId grainId, stri if (newEtag != null) { entry.ETag = newEtag; - // A request can arrive on a stale owner. Persist it here, but let the current owner load it. - if (RingRange.InRange(grainId)) - { - ReconcileLocalReminder(entry, _timeProvider.GetUtcNow().UtcDateTime); - } + var reconciliation = ReserveMutationReconciliation(grainId, reminderName); + await ReconcilePersistedReminder( + grainId, + reminderName, + ReminderEvents.LocalReminderStopReason.RemovedFromTable, + reconciliation); LogDebugRegisterReminder(entry, localTableSequence); @@ -271,18 +284,12 @@ public async Task UnregisterReminder(IGrainReminder reminder) if (success) { - var key = new ReminderIdentity(grainId, reminderName); - if (localReminders.TryGetValue(key, out var localRem)) - { - RequestLocalReminderRemoval(key, localRem, ReminderEvents.LocalReminderStopReason.Unregistered); - LogStoppedReminder(reminder); - if (logger.IsEnabled(LogLevel.Trace)) PrintReminders($"After removing {reminder}."); - } - else - { - AddLocalReminderTombstone(key, ReminderEvents.LocalReminderStopReason.Unregistered); - LogRemovedReminderFromTable(reminder); - } + var reconciliation = ReserveMutationReconciliation(grainId, reminderName); + await ReconcilePersistedReminder( + grainId, + reminderName, + ReminderEvents.LocalReminderStopReason.Unregistered, + reconciliation); ReminderEvents.EmitUnregistered(grainId, reminderName, Silo); } else @@ -348,9 +355,8 @@ private Task ReadAndUpdateReminders() var tasks = new List(); RemoveOutOfRangeReminders(tasks); - // Refreshes use even sequence values. Local writes use the following odd value, so they can supersede - // this snapshot without advancing the refresh generation. A newer refresh advances by two and causes - // all older refresh results to be discarded. + // Reserve a refresh generation. Local mutations advance this sequence and supersede ordinary snapshots; + // a later refresh advances it again and causes older ordinary refresh results to be discarded. var cachedSequence = localTableSequence += 2; var rangeSerialNumberCopy = RangeSerialNumber; LogTraceRingRange(RingRange, RangeSerialNumber, localReminders.Count); @@ -365,11 +371,37 @@ private Task ReadAndUpdateReminders() internal Task TestOnlyRefresh() { - var refreshTask = new Task(ReadAndUpdateReminders); + var refreshTask = new Task(() => ReadAndUpdateReminders()); Scheduler.QueueTask(refreshTask); return refreshTask.Unwrap(); } + internal async Task TestOnlyRegisterOrUpdateReminder( + GrainId grainId, + string reminderName, + TimeSpan dueTime, + TimeSpan period) + { + IGrainReminder? result = null; + await this.QueueTask(async () => result = await RegisterOrUpdateReminder(grainId, reminderName, dueTime, period)); + return result!; + } + + internal Task TestOnlyUnregisterReminder(IGrainReminder reminder) + => this.QueueTask(() => UnregisterReminder(reminder)); + + internal async Task TestOnlyGetLocalReminder(GrainId grainId, string reminderName) + { + ReminderEntry? result = null; + await this.QueueTask(() => + { + var key = new ReminderIdentity(grainId, reminderName); + result = localReminders.TryGetValue(key, out var reminder) ? reminder.Entry : null; + return Task.CompletedTask; + }); + return result; + } + private void RemoveOutOfRangeReminders(List removedReminderTasks) { CheckRuntimeContext(); @@ -549,7 +581,10 @@ private async Task DoInitialReadAndUpdateReminders() } } - private async Task ReadAndReconcileRange(ISingleRange range, int rangeSerialNumberCopy, long cachedSequence) + private async Task ReadAndReconcileRange( + ISingleRange range, + int rangeSerialNumberCopy, + long cachedSequence) { CheckRuntimeContext(); @@ -559,7 +594,7 @@ private async Task ReadAndReconcileRange(ISingleRange range, int rangeSerialNumb { // The read sequence was captured before any range read yielded. Local mutations which run while // storage is reading receive a later sequence and therefore win when this snapshot returns. - ReminderTableData? table = await reminderTable.ReadRows(range.Begin, range.End); // get all reminders, even the ones we already have + ReminderTableData? table = await reminderTable.ReadRows(range.Begin, range.End); if (cachedSequence < localTableSequence) { @@ -589,12 +624,41 @@ private async Task ReadAndReconcileRange(ISingleRange range, int rangeSerialNumb } } + foreach (var entry in table.Reminders) + { + if (range.InRange(entry.GrainId)) + { + remindersNotInTable.Remove(new(entry.GrainId, entry.ReminderName)); + } + } + + var stronglyConfirmedRows = new Dictionary(); + foreach (var batch in remindersNotInTable.BatchIEnumerable(MissingReminderPointReadConcurrency)) + { + var reads = batch.Select(async key => + (Key: key, Entry: await reminderTable.ReadRow(key.GrainId, key.ReminderName))); + foreach (var result in await Task.WhenAll(reads)) + { + stronglyConfirmedRows[result.Key] = result.Entry; + } + } + + if (cachedSequence < localTableSequence || rangeSerialNumberCopy < RangeSerialNumber) + { + return; + } + LogDebugReadRemindersFromTable(range, table.Reminders.Count, localTableSequence, cachedSequence); var tasks = new List(); // Use one timestamp for the entire snapshot so every row is evaluated against the same loading window. var now = _timeProvider.GetUtcNow().UtcDateTime; foreach (var entry in table.Reminders) { + if (!range.InRange(entry.GrainId)) + { + continue; + } + var key = new ReminderIdentity(entry.GrainId, entry.ReminderName); remindersNotInTable.Remove(key); ReconcileTableEntry(entry, cachedSequence, now, tasks); @@ -606,6 +670,12 @@ private async Task ReadAndReconcileRange(ISingleRange range, int rangeSerialNumb // return are no longer ours to schedule. foreach (var key in remindersNotInTable) { + if (stronglyConfirmedRows[key] is { } persistedEntry) + { + ReconcileTableEntry(persistedEntry, cachedSequence, now, tasks); + continue; + } + if (!localReminders.TryGetValue(key, out var reminder)) { continue; @@ -727,6 +797,196 @@ private void ReconcileLocalReminder(ReminderEntry entry, DateTime now) } } + private async Task ReconcilePersistedReminder( + GrainId grainId, + string reminderName, + ReminderEvents.LocalReminderStopReason removalReason, + (ReminderIdentity Key, MutationReconciliationState State, long Sequence) reconciliation) + { + try + { + if (!IsLatestMutationReconciliation(reconciliation)) + { + return; + } + + var owner = _ringProvider.GetPrimaryTargetSilo(grainId.GetUniformHashCode()); + if (owner is null) + { + return; + } + + if (!owner.Equals(Silo)) + { + if (!await TryNotifyReminderOwner(owner, grainId, reminderName, MutationNotificationMaxHops)) + { + LogWarningOwnerNotificationIncomplete(grainId, reminderName); + } + + return; + } + + var persistedEntry = await reminderTable.ReadRow(grainId, reminderName); + if (!IsLatestMutationReconciliation(reconciliation)) + { + return; + } + + owner = _ringProvider.GetPrimaryTargetSilo(grainId.GetUniformHashCode()); + if (owner is null) + { + return; + } + + if (!owner.Equals(Silo)) + { + if (!await TryNotifyReminderOwner(owner, grainId, reminderName, MutationNotificationMaxHops)) + { + LogWarningOwnerNotificationIncomplete(grainId, reminderName); + } + + return; + } + + if (persistedEntry is not null) + { + ReconcileLocalReminder(persistedEntry, _timeProvider.GetUtcNow().UtcDateTime); + return; + } + + if (localReminders.TryGetValue(reconciliation.Key, out var localReminder)) + { + RequestLocalReminderRemoval(reconciliation.Key, localReminder, removalReason); + } + else + { + AddLocalReminderTombstone(reconciliation.Key, removalReason); + } + } + + finally + { + CompleteMutationReconciliation(reconciliation); + } + } + + public async Task ReconcileReminder( + GrainId grainId, + string reminderName, + int remainingHops) + { + if (remainingHops < 0) + { + return false; + } + + var owner = _ringProvider.GetPrimaryTargetSilo(grainId.GetUniformHashCode()); + if (owner is null) + { + return false; + } + + if (!owner.Equals(Silo)) + { + return remainingHops > 0 + && await TryNotifyReminderOwner(owner, grainId, reminderName, remainingHops - 1); + } + + var reconciliation = ReserveMutationReconciliation(grainId, reminderName); + try + { + var persistedEntry = await reminderTable.ReadRow(grainId, reminderName); + if (!IsLatestMutationReconciliation(reconciliation)) + { + return true; + } + + owner = _ringProvider.GetPrimaryTargetSilo(grainId.GetUniformHashCode()); + if (owner is null) + { + return false; + } + + if (!owner.Equals(Silo)) + { + return remainingHops > 0 + && await TryNotifyReminderOwner(owner, grainId, reminderName, remainingHops - 1); + } + + if (persistedEntry is not null) + { + ReconcileLocalReminder(persistedEntry, _timeProvider.GetUtcNow().UtcDateTime); + } + else if (localReminders.TryGetValue(reconciliation.Key, out var localReminder)) + { + RequestLocalReminderRemoval( + reconciliation.Key, + localReminder, + ReminderEvents.LocalReminderStopReason.RemovedFromTable); + } + else + { + AddLocalReminderTombstone( + reconciliation.Key, + ReminderEvents.LocalReminderStopReason.RemovedFromTable); + } + + return true; + } + finally + { + CompleteMutationReconciliation(reconciliation); + } + } + + private async Task TryNotifyReminderOwner( + SiloAddress owner, + GrainId grainId, + string reminderName, + int remainingHops) + { + try + { + var serviceId = SystemTargetGrainId.CreateGrainServiceGrainId(_reminderServiceGrainType, owner); + var service = _grainFactory.GetSystemTarget(serviceId); + return await service.ReconcileReminder(grainId, reminderName, remainingHops); + } + catch (Exception) + { + return false; + } + } + + private (ReminderIdentity Key, MutationReconciliationState State, long Sequence) ReserveMutationReconciliation( + GrainId grainId, + string reminderName) + { + var key = new ReminderIdentity(grainId, reminderName); + if (!mutationReconciliations.TryGetValue(key, out var state)) + { + state = new(); + mutationReconciliations.Add(key, state); + } + + state.PendingCount++; + state.LatestSequence = ++localTableSequence; + return (key, state, state.LatestSequence); + } + + private static bool IsLatestMutationReconciliation( + (ReminderIdentity Key, MutationReconciliationState State, long Sequence) reconciliation) + => reconciliation.State.LatestSequence == reconciliation.Sequence; + + private void CompleteMutationReconciliation( + (ReminderIdentity Key, MutationReconciliationState State, long Sequence) reconciliation) + { + reconciliation.State.PendingCount--; + if (reconciliation.State.PendingCount == 0) + { + mutationReconciliations.Remove(reconciliation.Key); + } + } + private void AddOrUpdateLocalReminder(ReminderEntry entry) => AddOrUpdateLocalReminder(entry, GetLocalMutationSequence()); @@ -906,8 +1166,6 @@ void CheckRange() if (!RingRange.InRange(grainId)) { LogWarningNotResponsible(debugInfo, grainId, RingRange); - // For now, we still let the caller proceed without throwing an exception... the periodical mechanism will take care of reminders being registered at the wrong silo - // otherwise, we can either reject the request, or re-route the request } } } @@ -1440,6 +1698,13 @@ public override string ToString() private readonly record struct ScheduledTick(long ScheduleVersion, DateTime TickTime); } + private sealed class MutationReconciliationState + { + public long LatestSequence { get; set; } + + public int PendingCount { get; set; } + } + private readonly struct ReminderIdentity(GrainId grainId, string reminderName) : IEquatable { public readonly GrainId GrainId = grainId; @@ -1701,6 +1966,12 @@ private readonly struct ReminderIdentity(GrainId grainId, string reminderName) : )] private partial void LogWarningNotResponsible(string request, GrainId grainId, IRingRange range); + [LoggerMessage( + Level = LogLevel.Warning, + Message = "Could not confirm owner reconciliation after mutating reminder {ReminderName} for grain {GrainId}; discovery remains subject to provider convergence." + )] + private partial void LogWarningOwnerNotificationIncomplete(GrainId grainId, string reminderName); + [LoggerMessage( Level = LogLevel.Warning, Message = "Exception firing reminder \"{ReminderName}\" for grain {GrainId}" diff --git a/src/Orleans.Reminders/SystemTargetInterfaces/IReminderService.cs b/src/Orleans.Reminders/SystemTargetInterfaces/IReminderService.cs index d4aa771382e..dd3e5007ab4 100644 --- a/src/Orleans.Reminders/SystemTargetInterfaces/IReminderService.cs +++ b/src/Orleans.Reminders/SystemTargetInterfaces/IReminderService.cs @@ -40,6 +40,15 @@ public interface IReminderService : IGrainService /// A representing the operation. Task UnregisterReminder(IGrainReminder reminder); + /// + /// Reconciles a completed reminder mutation on the current owner, following a bounded number of topology redirects. + /// + /// The grain identity. + /// The reminder name. + /// The remaining topology redirects. + /// when an owner reconciled the mutation; otherwise . + Task ReconcileReminder(GrainId grainId, string reminderName, int remainingHops); + /// /// Gets the reminder registered to the specified grain with the provided name. /// diff --git a/src/api/AWS/Orleans.Reminders.DynamoDB/Orleans.Reminders.DynamoDB.cs b/src/api/AWS/Orleans.Reminders.DynamoDB/Orleans.Reminders.DynamoDB.cs index 0a60a970f8f..e3c7f8bbd0f 100644 --- a/src/api/AWS/Orleans.Reminders.DynamoDB/Orleans.Reminders.DynamoDB.cs +++ b/src/api/AWS/Orleans.Reminders.DynamoDB/Orleans.Reminders.DynamoDB.cs @@ -8,18 +8,33 @@ //------------------------------------------------------------------------------ namespace Orleans.Configuration { + public enum DynamoDBReminderTableMode + { + Legacy = 0, + Migrate = 1, + V2 = 2, + Rollback = 3, + V2Only = 4, + } + public partial class DynamoDBReminderStorageOptions : Reminders.DynamoDB.DynamoDBClientOptions { public bool CreateIfNotExists { get { throw null; } set { } } public int ReadCapacityUnits { get { throw null; } set { } } + public int MigrationPageSize { get { throw null; } set { } } + public string TableName { get { throw null; } set { } } + public DynamoDBReminderTableMode TableMode { get { throw null; } set { } } + public bool UpdateIfExists { get { throw null; } set { } } public bool UseProvisionedThroughput { get { throw null; } set { } } + public string? V2TableName { get { throw null; } set { } } + public int WriteCapacityUnits { get { throw null; } set { } } } diff --git a/src/api/Orleans.Reminders/Orleans.Reminders.cs b/src/api/Orleans.Reminders/Orleans.Reminders.cs index de67432dca4..b590c709b2b 100644 --- a/src/api/Orleans.Reminders/Orleans.Reminders.cs +++ b/src/api/Orleans.Reminders/Orleans.Reminders.cs @@ -36,6 +36,7 @@ public partial interface IReminderService : Services.IGrainService, ISystemTarge { System.Threading.Tasks.Task GetReminder(Runtime.GrainId grainId, string reminderName); System.Threading.Tasks.Task> GetReminders(Runtime.GrainId grainId); + System.Threading.Tasks.Task ReconcileReminder(Runtime.GrainId grainId, string reminderName, int remainingHops); System.Threading.Tasks.Task RegisterOrUpdateReminder(Runtime.GrainId grainId, string reminderName, System.TimeSpan dueTime, System.TimeSpan period); System.Threading.Tasks.Task Start(); System.Threading.Tasks.Task Stop(); diff --git a/test/Extensions/Orleans.AWS.Tests/Reminder/DynamoDBReminderMigrationTests.cs b/test/Extensions/Orleans.AWS.Tests/Reminder/DynamoDBReminderMigrationTests.cs new file mode 100644 index 00000000000..7280c5951b6 --- /dev/null +++ b/test/Extensions/Orleans.AWS.Tests/Reminder/DynamoDBReminderMigrationTests.cs @@ -0,0 +1,918 @@ +using Amazon.DynamoDBv2; +using Amazon.DynamoDBv2.Model; +using Amazon.Runtime; +using AWSUtils.Tests.StorageTests; +using Microsoft.Extensions.Logging.Abstractions; +using Microsoft.Extensions.Options; +using Orleans.Configuration; +using Orleans.Reminders.DynamoDB; +using Orleans.Runtime; +using System.Collections.Immutable; +using TestExtensions; +using Xunit; + +namespace AWSUtils.Tests.RemindersTest; + +[TestCategory("Reminders"), TestCategory("AWS"), TestCategory("DynamoDb")] +[Collection(TestEnvironmentFixture.DefaultCollection)] +[TestSuite("Functional")] +[TestProvider("DynamoDB")] +[TestArea("Reminders")] +public sealed class DynamoDBReminderMigrationTests +{ + [Fact] + public void V2KeyEncoding_IsSortableUnambiguousAndHasStableBuckets() + { + var grain = GrainId.Create("type/#", "key_+/"); + var otherGrain = GrainId.Create("type/#", "key_+"); + + var first = DynamoDBReminderTable.GetV2SortKey(0, grain, "name/#_+"); + var last = DynamoDBReminderTable.GetV2SortKey(uint.MaxValue, grain, "name/#_+"); + + Assert.StartsWith("R#00000000#", first, StringComparison.Ordinal); + Assert.StartsWith("R#FFFFFFFF#", last, StringComparison.Ordinal); + Assert.DoesNotContain("/", first, StringComparison.Ordinal); + Assert.DoesNotContain("+", first, StringComparison.Ordinal); + Assert.NotEqual(first, DynamoDBReminderTable.GetV2SortKey(0, otherGrain, "name/#_+")); + Assert.NotEqual(first, DynamoDBReminderTable.GetV2SortKey(0, grain, "name/#_")); + Assert.True(string.CompareOrdinal(first, last) < 0); + var longNameKey = DynamoDBReminderTable.GetV2SortKey(1, grain, new string('x', 1_500)); + Assert.True(System.Text.Encoding.UTF8.GetByteCount(longNameKey) < 1_024); + Assert.NotEqual(longNameKey, DynamoDBReminderTable.GetV2SortKey(1, grain, new string('x', 1_499) + "y")); + Assert.Equal( + DynamoDBReminderTable.GetV2PartitionKey("service/#", 0), + DynamoDBReminderTable.GetV2PartitionKey("service/#", DynamoDBReminderTable.V2BucketCount)); + Assert.NotEqual( + DynamoDBReminderTable.GetV2PartitionKey("service/#", 0), + DynamoDBReminderTable.GetV2PartitionKey("service/#", 1)); + + Assert.Equal(("R#00000000#", "R#FFFFFFFF#~"), DynamoDBReminderTable.GetV2RangeBounds(0, uint.MaxValue)); + Assert.Equal(("R#00000001#", "R#00000001#~"), DynamoDBReminderTable.GetV2RangeBounds(1, 1)); + } + + [Fact] + public void MigrationOptions_ParseConnectionStringIncludesCustomV2Settings() + { + var options = new DynamoDBReminderStorageOptions(); + + options.ParseConnectionString( + "Service=us-east-2;TableName=custom-v1;V2TableName=custom-v2;TableMode=Migrate;MigrationPageSize=37"); + + Assert.Equal("us-east-2", options.Service); + Assert.Equal("custom-v1", options.TableName); + Assert.Equal("custom-v2", options.V2TableName); + Assert.Equal(DynamoDBReminderTableMode.Migrate, options.TableMode); + Assert.Equal(37, options.MigrationPageSize); + } + + [Theory] + [InlineData(0)] + [InlineData(-1)] + public async Task MigrationOptions_RejectNonpositivePageSize(int pageSize) + { + var table = CreateTable(NewTableName(), "invalid-options", DynamoDBReminderTableMode.Legacy, pageSize); + + var exception = await Assert.ThrowsAsync( + () => table.StartAsync(TestContext.Current.CancellationToken)); + + Assert.Equal("MigrationPageSize", exception.ParamName); + } + + [Fact] + public async Task Init_MigrateModeCreatesV2BaseTableSchema() + { + EnsureDynamoDb(); + var tableName = NewTableName(); + using var client = CreateClient(); + var table = CreateTable(tableName, "schema", DynamoDBReminderTableMode.Migrate); + try + { + await table.StartAsync(TestContext.Current.CancellationToken); + var description = (await client.DescribeTableAsync($"{tableName}-v2", TestContext.Current.CancellationToken)).Table; + + Assert.Collection( + description.KeySchema, + item => + { + Assert.Equal(DynamoDBReminderTable.V2PartitionKeyName, item.AttributeName); + Assert.Equal(KeyType.HASH, item.KeyType); + }, + item => + { + Assert.Equal(DynamoDBReminderTable.V2SortKeyName, item.AttributeName); + Assert.Equal(KeyType.RANGE, item.KeyType); + }); + Assert.True(description.GlobalSecondaryIndexes is null or { Count: 0 }); + Assert.Equal(2, description.AttributeDefinitions.Count); + } + finally + { + await StopAndDelete(client, tableName, table); + } + } + + [Fact] + public async Task LegacyDiscovery_StrongPointValidationReturnsCurrentRowsAndDropsDeletes() + { + EnsureDynamoDb(); + var tableName = NewTableName(); + const string serviceId = "legacy-candidates"; + using var client = CreateClient(); + IReadOnlyList discovery = []; + var table = CreateTable( + tableName, + serviceId, + DynamoDBReminderTableMode.Legacy, + hooks: new() { LegacyDiscoveryResults = _ => discovery }); + try + { + await table.StartAsync(TestContext.Current.CancellationToken); + var entry = Entry(1); + var firstEtag = await table.UpsertRow(entry); + discovery = [Clone(entry)]; + + entry.StartAt = entry.StartAt.AddHours(3); + entry.Period = TimeSpan.FromHours(2); + var secondEtag = await table.UpsertRow(entry); + var rangeResult = Assert.Single((await table.ReadRows(0, 0)).Reminders); + var grainResult = Assert.Single((await table.ReadRows(entry.GrainId)).Reminders); + Assert.Equal(secondEtag, rangeResult.ETag); + Assert.Equal(entry.StartAt, rangeResult.StartAt); + Assert.Equal(entry.Period, rangeResult.Period); + Assert.Equal(secondEtag, grainResult.ETag); + + Assert.True(await table.RemoveRow(entry.GrainId, entry.ReminderName, secondEtag!)); + Assert.Empty((await table.ReadRows(0, 0)).Reminders); + Assert.Empty((await table.ReadRows(entry.GrainId)).Reminders); + Assert.NotEqual(firstEtag, secondEtag); + } + finally + { + await StopAndDelete(client, tableName, table); + } + } + + [Fact] + public async Task LegacyMissingDiscoveryCandidateRemainsUndiscoverableWithoutSchemaChange() + { + EnsureDynamoDb(); + var tableName = NewTableName(); + using var client = CreateClient(); + var table = CreateTable( + tableName, + "legacy-missing", + DynamoDBReminderTableMode.Legacy, + hooks: new() { LegacyDiscoveryResults = _ => [] }); + try + { + await table.StartAsync(TestContext.Current.CancellationToken); + var entry = Entry(1); + await table.UpsertRow(entry); + + Assert.Empty((await table.ReadRows(0, 0)).Reminders); + Assert.Empty((await table.ReadRows(entry.GrainId)).Reminders); + var pointRead = Assert.IsType(await table.ReadRow(entry.GrainId, entry.ReminderName)); + Assert.Equal(entry.GrainId, pointRead.GrainId); + Assert.Equal(entry.ReminderName, pointRead.ReminderName); + Assert.Equal(entry.ETag, pointRead.ETag); + } + finally + { + await StopAndDelete(client, tableName, table); + } + } + + [Fact] + public async Task Migration_BackfillsPagesResumesAndPreservesLegacyETags() + { + EnsureDynamoDb(); + var tableName = NewTableName(); + const string serviceId = "resume"; + using var client = CreateClient(); + var legacy = CreateTable(tableName, serviceId, DynamoDBReminderTableMode.Legacy); + DynamoDBReminderTable? interrupted = null; + DynamoDBReminderTable? resumed = null; + try + { + await legacy.StartAsync(TestContext.Current.CancellationToken); + var entries = Enumerable.Range(0, 8).Select(index => Entry(index)).ToArray(); + entries[^1].ReminderName = new string('x', 1_500); + foreach (var entry in entries) + { + await legacy.UpsertRow(entry); + } + + var pages = 0; + interrupted = CreateTable( + tableName, + serviceId, + DynamoDBReminderTableMode.Migrate, + pageSize: 2, + hooks: new() + { + AfterPageCheckpoint = () => Interlocked.Increment(ref pages) == 1 + ? Task.FromException(new InjectedMigrationException()) + : Task.CompletedTask, + }); + await Assert.ThrowsAsync( + () => interrupted.StartAsync(TestContext.Current.CancellationToken)); + + var state = await ReadState(client, tableName, serviceId); + Assert.Equal("Backfilling", state["MigrationStatus"].S); + Assert.True(state.ContainsKey("CheckpointReminderId")); + Assert.True(state.ContainsKey("CheckpointGrainHash")); + + resumed = CreateTable(tableName, serviceId, DynamoDBReminderTableMode.Migrate, pageSize: 2); + await resumed.StartAsync(TestContext.Current.CancellationToken); + + state = await ReadState(client, tableName, serviceId); + Assert.Equal("Ready", state["MigrationStatus"].S); + Assert.False(state.ContainsKey("CheckpointReminderId")); + var v2Items = await ReadV2Items(client, tableName, serviceId); + Assert.Equal(entries.Length, v2Items.Count); + Assert.Equal( + entries.Select(static entry => entry.ETag).OrderBy(static value => value, StringComparer.Ordinal), + v2Items.Select(static item => Scalar(item["ETag"])).OrderBy(static value => value, StringComparer.Ordinal)); + foreach (var entry in entries) + { + var actual = Assert.Single(v2Items, item => item["GrainReference"].S == entry.GrainId.ToString()); + Assert.Equal(entry.StartAt, DateTime.Parse(actual["StartTime"].S)); + Assert.Equal(entry.Period, TimeSpan.Parse(actual["Period"].S)); + } + } + finally + { + await StopAndDelete(client, tableName, legacy, interrupted, resumed); + } + } + + [Fact] + public async Task Migration_ConcurrentLegacyUpdateAndDeleteCannotResurrectStaleRows() + { + EnsureDynamoDb(); + var tableName = NewTableName(); + const string serviceId = "concurrent"; + using var client = CreateClient(); + var legacy = CreateTable(tableName, serviceId, DynamoDBReminderTableMode.Legacy); + DynamoDBReminderTable? migration = null; + try + { + await legacy.StartAsync(TestContext.Current.CancellationToken); + var updated = Entry(1); + var deleted = Entry(2); + var staleDelete = Entry(3); + await legacy.UpsertRow(updated); + await legacy.UpsertRow(deleted); + await legacy.UpsertRow(staleDelete); + + var pageRead = new TaskCompletionSource(TaskCreationOptions.RunContinuationsAsynchronously); + var release = new TaskCompletionSource(TaskCreationOptions.RunContinuationsAsynchronously); + var invoked = 0; + migration = CreateTable( + tableName, + serviceId, + DynamoDBReminderTableMode.Migrate, + pageSize: 100, + hooks: new() + { + AfterLegacyPageRead = async () => + { + if (Interlocked.Exchange(ref invoked, 1) == 0) + { + pageRead.SetResult(); + await release.Task; + } + }, + }); + + var migrationTask = migration.StartAsync(TestContext.Current.CancellationToken); + await pageRead.Task.WaitAsync(TestContext.Current.CancellationToken); + var deletedEtag = deleted.ETag!; + updated.StartAt = updated.StartAt.AddDays(3); + var updatedEtag = await legacy.UpsertRow(updated); + staleDelete.Period = TimeSpan.FromDays(2); + var staleDeleteEtag = await legacy.UpsertRow(staleDelete); + Assert.True(await migration.RemoveRow(staleDelete.GrainId, staleDelete.ReminderName, staleDeleteEtag!)); + Assert.True(await legacy.RemoveRow(deleted.GrainId, deleted.ReminderName, deletedEtag)); + release.SetResult(); + await migrationTask; + + var v2Items = await ReadV2Items(client, tableName, serviceId); + var actual = Assert.Single(v2Items); + Assert.Equal(updated.GrainId.ToString(), actual["GrainReference"].S); + Assert.Equal(updatedEtag, Scalar(actual["ETag"])); + Assert.Equal(updated.StartAt, DateTime.Parse(actual["StartTime"].S)); + } + finally + { + await StopAndDelete(client, tableName, legacy, migration); + } + } + + [Fact] + public async Task Migration_LeaseContentionHasOneOwnerAndContenderRemainsDualWriteCapable() + { + EnsureDynamoDb(); + var tableName = NewTableName(); + const string serviceId = "lease"; + using var client = CreateClient(); + var legacy = CreateTable(tableName, serviceId, DynamoDBReminderTableMode.Legacy); + DynamoDBReminderTable? owner = null; + DynamoDBReminderTable? contender = null; + try + { + await legacy.StartAsync(TestContext.Current.CancellationToken); + await legacy.UpsertRow(Entry(1)); + var pageRead = new TaskCompletionSource(TaskCreationOptions.RunContinuationsAsynchronously); + var release = new TaskCompletionSource(TaskCreationOptions.RunContinuationsAsynchronously); + owner = CreateTable( + tableName, + serviceId, + DynamoDBReminderTableMode.Migrate, + hooks: new() + { + AfterLegacyPageRead = async () => + { + pageRead.SetResult(); + await release.Task; + }, + }); + var ownerTask = owner.StartAsync(TestContext.Current.CancellationToken); + await pageRead.Task.WaitAsync(TestContext.Current.CancellationToken); + + contender = CreateTable(tableName, serviceId, DynamoDBReminderTableMode.Migrate); + await contender.StartAsync(TestContext.Current.CancellationToken); + var contenderEntry = Entry(2); + var etag = await contender.UpsertRow(contenderEntry); + Assert.NotNull(etag); + + release.SetResult(); + await ownerTask; + Assert.Equal(2, (await ReadV2Items(client, tableName, serviceId)).Count); + } + finally + { + await StopAndDelete(client, tableName, legacy, owner, contender); + } + } + + [Fact] + public async Task Migration_VerificationFailureIsPersistedAndPreventsCutover() + { + EnsureDynamoDb(); + var tableName = NewTableName(); + const string serviceId = "verification"; + using var client = CreateClient(); + var legacy = CreateTable(tableName, serviceId, DynamoDBReminderTableMode.Legacy); + DynamoDBReminderTable? migration = null; + try + { + await legacy.StartAsync(TestContext.Current.CancellationToken); + var entry = Entry(1); + await legacy.UpsertRow(entry); + migration = CreateTable( + tableName, + serviceId, + DynamoDBReminderTableMode.Migrate, + hooks: new() + { + BeforeVerification = async () => + { + await client.DeleteItemAsync( + new() + { + TableName = $"{tableName}-v2", + Key = V2Key(serviceId, entry), + }, + TestContext.Current.CancellationToken); + }, + }); + + var exception = await Assert.ThrowsAsync( + () => migration.StartAsync(TestContext.Current.CancellationToken)); + Assert.Contains("verification failed", exception.Message, StringComparison.OrdinalIgnoreCase); + var state = await ReadState(client, tableName, serviceId); + Assert.Equal("VerificationFailed", state["MigrationStatus"].S); + Assert.Equal("1", state["SourceCount"].N); + Assert.Equal("0", state["TargetCount"].N); + Assert.NotNull(await legacy.ReadRow(entry.GrainId, entry.ReminderName)); + } + finally + { + await StopAndDelete(client, tableName, legacy, migration); + } + } + + [Fact] + public async Task Migration_ExpiredLeaseCannotPublishCutover() + { + EnsureDynamoDb(); + var tableName = NewTableName(); + const string serviceId = "expired-lease"; + using var client = CreateClient(); + var clock = new ManualTimeProvider(new DateTimeOffset(2026, 8, 28, 10, 0, 0, TimeSpan.Zero)); + var cluster = CompatibleCluster(); + var legacy = CreateTable(tableName, serviceId, DynamoDBReminderTableMode.Legacy); + var table = CreateTable( + tableName, + serviceId, + DynamoDBReminderTableMode.V2, + hooks: new() + { + AfterLegacyPageRead = () => + { + clock.Advance(TimeSpan.FromMinutes(3)); + return Task.CompletedTask; + }, + }, + membership: cluster.Membership, + local: cluster.Local, + timeProvider: clock); + try + { + await legacy.StartAsync(TestContext.Current.CancellationToken); + await legacy.UpsertRow(Entry(1)); + var exception = await Assert.ThrowsAsync( + () => table.StartAsync(TestContext.Current.CancellationToken)); + Assert.Contains("lost while copying", exception.Message, StringComparison.Ordinal); + Assert.NotEqual("Cutover", (await ReadState(client, tableName, serviceId))["MigrationStatus"].S); + } + finally + { + await StopAndDelete(client, tableName, legacy, table); + } + } + + [Fact] + public async Task MixedVersionPrepareCutoverAndRollbackPreserveStrongReadsAndETags() + { + EnsureDynamoDb(); + var tableName = NewTableName(); + const string serviceId = "cutover"; + using var client = CreateClient(); + var legacy = CreateTable(tableName, serviceId, DynamoDBReminderTableMode.Legacy); + DynamoDBReminderTable? prepare = null; + DynamoDBReminderTable? cutover = null; + DynamoDBReminderTable? failedRollback = null; + DynamoDBReminderTable? rollback = null; + try + { + await legacy.StartAsync(TestContext.Current.CancellationToken); + var oldEntry = Entry(1); + await legacy.UpsertRow(oldEntry); + prepare = CreateTable(tableName, serviceId, DynamoDBReminderTableMode.Migrate); + await prepare.StartAsync(TestContext.Current.CancellationToken); + + var oldBinaryEntry = Entry(2); + await legacy.UpsertRow(oldBinaryEntry); + var cluster = CompatibleCluster(); + cutover = CreateTable( + tableName, + serviceId, + DynamoDBReminderTableMode.V2, + membership: cluster.Membership, + local: cluster.Local); + await cutover.StartAsync(TestContext.Current.CancellationToken); + Assert.Equal("Cutover", (await ReadState(client, tableName, serviceId))["MigrationStatus"].S); + + var current = Entry(3); + var firstEtag = await cutover.UpsertRow(current); + var point = Assert.IsType(await cutover.ReadRow(current.GrainId, current.ReminderName)); + Assert.Equal(firstEtag, point.ETag); + Assert.Contains((await cutover.ReadRows(current.GrainId)).Reminders, item => item.ETag == firstEtag); + Assert.Contains((await cutover.ReadRows(0, 0)).Reminders, item => item.ETag == firstEtag); + + current.Period = TimeSpan.FromMinutes(99); + var secondEtag = await cutover.UpsertRow(current); + Assert.NotEqual(firstEtag, secondEtag); + Assert.False(await cutover.RemoveRow(current.GrainId, current.ReminderName, firstEtag!)); + Assert.True(await cutover.RemoveRow(current.GrainId, current.ReminderName, secondEtag!)); + Assert.Null(await cutover.ReadRow(current.GrainId, current.ReminderName)); + Assert.DoesNotContain((await cutover.ReadRows(current.GrainId)).Reminders, item => item.ReminderName == current.ReminderName); + Assert.DoesNotContain((await cutover.ReadRows(0, 0)).Reminders, item => item.ReminderName == current.ReminderName); + + failedRollback = CreateTable( + tableName, + serviceId, + DynamoDBReminderTableMode.Rollback, + hooks: new() + { + BeforeVerification = async () => + { + await client.DeleteItemAsync( + new() + { + TableName = $"{tableName}-v2", + Key = V2Key(serviceId, oldEntry), + }, + TestContext.Current.CancellationToken); + }, + }, + membership: cluster.Membership, + local: cluster.Local); + await Assert.ThrowsAsync( + () => failedRollback.StartAsync(TestContext.Current.CancellationToken)); + Assert.Equal("Cutover", (await ReadState(client, tableName, serviceId))["MigrationStatus"].S); + + rollback = CreateTable( + tableName, + serviceId, + DynamoDBReminderTableMode.Rollback, + membership: cluster.Membership, + local: cluster.Local); + await rollback.StartAsync(TestContext.Current.CancellationToken); + Assert.Equal("RolledBack", (await ReadState(client, tableName, serviceId))["MigrationStatus"].S); + Assert.NotNull(await rollback.ReadRow(oldEntry.GrainId, oldEntry.ReminderName)); + Assert.NotNull(await rollback.ReadRow(oldBinaryEntry.GrainId, oldBinaryEntry.ReminderName)); + } + finally + { + await StopAndDelete(client, tableName, legacy, prepare, cutover, failedRollback, rollback); + } + } + + [Fact] + public async Task V2RangeReads_AreBeginExclusiveEndInclusiveForNormalAndWrapRanges() + { + EnsureDynamoDb(); + var tableName = NewTableName(); + const string serviceId = "ranges"; + using var client = CreateClient(); + var prepare = CreateTable(tableName, serviceId, DynamoDBReminderTableMode.Migrate); + DynamoDBReminderTable? cutover = null; + try + { + await prepare.StartAsync(TestContext.Current.CancellationToken); + var entries = Enumerable.Range(0, 8) + .Select(Entry) + .OrderBy(static entry => entry.GrainId.GetUniformHashCode()) + .ToArray(); + foreach (var entry in entries) + { + await prepare.UpsertRow(entry); + } + + var cluster = CompatibleCluster(); + cutover = CreateTable( + tableName, + serviceId, + DynamoDBReminderTableMode.V2, + membership: cluster.Membership, + local: cluster.Local); + await cutover.StartAsync(TestContext.Current.CancellationToken); + + var normal = await cutover.ReadRows( + entries[1].GrainId.GetUniformHashCode(), + entries[5].GrainId.GetUniformHashCode()); + Assert.Equal( + entries[2..6].Select(static entry => entry.GrainId.ToString()).OrderBy(static id => id, StringComparer.Ordinal), + normal.Reminders.Select(static entry => entry.GrainId.ToString()).OrderBy(static id => id, StringComparer.Ordinal)); + + var wrap = await cutover.ReadRows( + entries[5].GrainId.GetUniformHashCode(), + entries[1].GrainId.GetUniformHashCode()); + Assert.Equal( + entries[6..].Concat(entries[..2]).Select(static entry => entry.GrainId.ToString()).OrderBy(static id => id, StringComparer.Ordinal), + wrap.Reminders.Select(static entry => entry.GrainId.ToString()).OrderBy(static id => id, StringComparer.Ordinal)); + + var fullRing = await cutover.ReadRows(entries[3].GrainId.GetUniformHashCode(), entries[3].GrainId.GetUniformHashCode()); + Assert.Equal(entries.Length, fullRing.Reminders.Count); + } + finally + { + await StopAndDelete(client, tableName, prepare, cutover); + } + } + + [Theory] + [InlineData(SiloStatus.Created)] + [InlineData(SiloStatus.Joining)] + [InlineData(SiloStatus.Active)] + public async Task Cutover_FailsClosedWhenANonterminalSiloHasNoCompatibilityMarker(SiloStatus incompatibleStatus) + { + EnsureDynamoDb(); + var tableName = NewTableName(); + using var client = CreateClient(); + var cluster = CompatibleCluster(incompatibleStatus); + var table = CreateTable( + tableName, + "mixed", + DynamoDBReminderTableMode.V2, + membership: cluster.Membership, + local: cluster.Local); + try + { + var exception = await Assert.ThrowsAsync( + () => table.StartAsync(TestContext.Current.CancellationToken)); + Assert.Contains("did not publish V2 compatibility markers", exception.Message, StringComparison.Ordinal); + } + finally + { + await StopAndDelete(client, tableName, table); + } + } + + [Fact] + public async Task V2Only_RetiresLegacyTableAndExistingV2InstancesFollowTheFence() + { + EnsureDynamoDb(); + var tableName = NewTableName(); + const string serviceId = "retirement"; + using var client = CreateClient(); + var prepare = CreateTable(tableName, serviceId, DynamoDBReminderTableMode.Migrate); + DynamoDBReminderTable? cutover = null; + DynamoDBReminderTable? retirement = null; + DynamoDBReminderTable? restarted = null; + try + { + await prepare.StartAsync(TestContext.Current.CancellationToken); + var retained = Entry(1); + await prepare.UpsertRow(retained); + var cluster = CompatibleCluster(); + cutover = CreateTable( + tableName, + serviceId, + DynamoDBReminderTableMode.V2, + membership: cluster.Membership, + local: cluster.Local); + await cutover.StartAsync(TestContext.Current.CancellationToken); + + retirement = CreateTable( + tableName, + serviceId, + DynamoDBReminderTableMode.V2Only, + membership: cluster.Membership, + local: cluster.Local); + await retirement.StartAsync(TestContext.Current.CancellationToken); + Assert.Equal("Retired", (await ReadState(client, tableName, serviceId))["MigrationStatus"].S); + + await client.DeleteTableAsync(new DeleteTableRequest { TableName = tableName }, TestContext.Current.CancellationToken); + + var afterRetirement = Entry(2); + var etag = await cutover.UpsertRow(afterRetirement); + Assert.Equal(etag, (await cutover.ReadRow(afterRetirement.GrainId, afterRetirement.ReminderName))?.ETag); + + restarted = CreateTable( + tableName, + serviceId, + DynamoDBReminderTableMode.V2, + membership: cluster.Membership, + local: cluster.Local); + await restarted.StartAsync(TestContext.Current.CancellationToken); + Assert.NotNull(await restarted.ReadRow(retained.GrainId, retained.ReminderName)); + await Assert.ThrowsAsync( + () => client.DescribeTableAsync(tableName, TestContext.Current.CancellationToken)); + Assert.True(await restarted.RemoveRow(afterRetirement.GrainId, afterRetirement.ReminderName, etag!)); + } + finally + { + await StopAndDelete(client, tableName, prepare, cutover, retirement, restarted); + } + } + + [Fact] + public async Task Migration_IsServiceIsolatedAndBackfillsAllPaginatedRows() + { + EnsureDynamoDb(); + var tableName = NewTableName(); + using var client = CreateClient(); + var serviceA = CreateTable(tableName, "service-a", DynamoDBReminderTableMode.Legacy); + var serviceB = CreateTable(tableName, "service-b", DynamoDBReminderTableMode.Legacy); + DynamoDBReminderTable? migrationA = null; + try + { + await serviceA.StartAsync(TestContext.Current.CancellationToken); + await serviceB.StartAsync(TestContext.Current.CancellationToken); + for (var index = 0; index < 12; index++) + { + await serviceA.UpsertRow(Entry(index)); + await serviceB.UpsertRow(Entry(index)); + } + + var checkpoints = 0; + migrationA = CreateTable( + tableName, + "service-a", + DynamoDBReminderTableMode.Migrate, + pageSize: 1, + hooks: new() { AfterPageCheckpoint = () => { checkpoints++; return Task.CompletedTask; } }); + await migrationA.StartAsync(TestContext.Current.CancellationToken); + + Assert.True(checkpoints >= 12); + Assert.Equal(12, (await ReadV2Items(client, tableName, "service-a")).Count); + Assert.Empty(await ReadV2Items(client, tableName, "service-b")); + } + finally + { + await StopAndDelete(client, tableName, serviceA, serviceB, migrationA); + } + } + + private static DynamoDBReminderTable CreateTable( + string tableName, + string serviceId, + DynamoDBReminderTableMode mode, + int pageSize = 100, + DynamoDBReminderMigrationTestHooks? hooks = null, + IClusterMembershipService? membership = null, + ILocalSiloDetails? local = null, + TimeProvider? timeProvider = null) + { + var options = new DynamoDBReminderStorageOptions + { + Service = AWSTestConstants.DynamoDbService, + AccessKey = AWSTestConstants.DynamoDbAccessKey, + SecretKey = AWSTestConstants.DynamoDbSecretKey, + TableName = tableName, + TableMode = mode, + MigrationPageSize = pageSize, + CreateIfNotExists = true, + UpdateIfExists = false, + UseProvisionedThroughput = false, + }; + return new( + NullLoggerFactory.Instance, + Options.Create(new ClusterOptions { ClusterId = serviceId, ServiceId = serviceId }), + Options.Create(options), + membership, + local, + timeProvider ?? TimeProvider.System, + hooks); + } + + private static ReminderEntry Entry(int index) + => new() + { + GrainId = GrainId.Create("migration", $"grain-{index:D4}"), + ReminderName = $"reminder/#_{index:D4}", + StartAt = new DateTime(2026, 8, 28, 1, 2, 3, DateTimeKind.Utc).AddMinutes(index), + Period = TimeSpan.FromMinutes(index + 1), + }; + + private static ReminderEntry Clone(ReminderEntry entry) + => new() + { + GrainId = entry.GrainId, + ReminderName = entry.ReminderName, + StartAt = entry.StartAt, + Period = entry.Period, + ETag = entry.ETag, + }; + + private static Dictionary V2Key(string serviceId, ReminderEntry entry) + { + var hash = entry.GrainId.GetUniformHashCode(); + return new() + { + [DynamoDBReminderTable.V2PartitionKeyName] = new(DynamoDBReminderTable.GetV2PartitionKey(serviceId, hash)), + [DynamoDBReminderTable.V2SortKeyName] = new(DynamoDBReminderTable.GetV2SortKey(hash, entry.GrainId, entry.ReminderName)), + }; + } + + private static async Task> ReadState( + AmazonDynamoDBClient client, + string tableName, + string serviceId) + { + var response = await client.ScanAsync( + new() + { + TableName = $"{tableName}-v2", + ConsistentRead = true, + FilterExpression = "MigrationStatus = :status OR attribute_exists(MigrationStatus)", + ExpressionAttributeValues = new() { [":status"] = new("unused") }, + }, + TestContext.Current.CancellationToken); + return Assert.Single(response.Items, item => item["SortKey"].S == "STATE" && item["PartitionKey"].S.StartsWith("M#", StringComparison.Ordinal)); + } + + private static async Task>> ReadV2Items( + AmazonDynamoDBClient client, + string tableName, + string serviceId) + { + var result = new List>(); + for (uint bucket = 0; bucket < DynamoDBReminderTable.V2BucketCount; bucket++) + { + var partition = DynamoDBReminderTable.GetV2PartitionKey(serviceId, bucket); + var response = await client.QueryAsync( + new() + { + TableName = $"{tableName}-v2", + ConsistentRead = true, + KeyConditionExpression = "PartitionKey = :partition", + ExpressionAttributeValues = new() { [":partition"] = new(partition) }, + }, + TestContext.Current.CancellationToken); + result.AddRange(response.Items); + } + + return result; + } + + private static string Scalar(AttributeValue value) => value.S ?? value.N; + + private static (IClusterMembershipService Membership, ILocalSiloDetails Local) CompatibleCluster(SiloStatus? incompatibleStatus = null) + { + var localAddress = SiloAddress.FromParsableString("127.0.0.1:21111@100"); + var members = ImmutableDictionary.Empty + .Add(localAddress, new(localAddress, SiloStatus.Active, "local")); + if (incompatibleStatus is { } status) + { + var oldAddress = SiloAddress.FromParsableString("127.0.0.1:21112@101"); + members = members.Add(oldAddress, new(oldAddress, status, "old")); + } + + return (new TestMembershipService(new(members, new(1))), new TestLocalSiloDetails(localAddress)); + } + + private static string NewTableName() => $"OrleansReminders-{Guid.NewGuid():N}"; + + private static void EnsureDynamoDb() + { + if (!AWSTestConstants.IsDynamoDbAvailable) + { + throw Xunit.Sdk.SkipException.ForSkip("Unable to connect to AWS DynamoDB simulator"); + } + } + + private static AmazonDynamoDBClient CreateClient() + { + var service = AWSTestConstants.DynamoDbService; + if (service.StartsWith("http://", StringComparison.OrdinalIgnoreCase) + || service.StartsWith("https://", StringComparison.OrdinalIgnoreCase)) + { + return new(new BasicAWSCredentials("dummy", "dummyKey"), new AmazonDynamoDBConfig { ServiceURL = service }); + } + + var config = new AmazonDynamoDBConfig { RegionEndpoint = Amazon.RegionEndpoint.GetBySystemName(service) }; + return string.IsNullOrEmpty(AWSTestConstants.DynamoDbAccessKey) + || string.IsNullOrEmpty(AWSTestConstants.DynamoDbSecretKey) + ? new(config) + : new( + new BasicAWSCredentials(AWSTestConstants.DynamoDbAccessKey, AWSTestConstants.DynamoDbSecretKey), + config); + } + + private static async Task StopAndDelete(AmazonDynamoDBClient client, string tableName, params DynamoDBReminderTable?[] tables) + { + foreach (var table in tables) + { + if (table is not null) + { + await table.StopAsync(); + } + + } + + foreach (var name in new[] { tableName, $"{tableName}-v2" }) + { + try + { + await client.DeleteTableAsync(new DeleteTableRequest { TableName = name }); + } + catch (ResourceNotFoundException) + { + } + } + } + + private sealed class InjectedMigrationException : Exception + { + } + + private sealed class TestLocalSiloDetails(SiloAddress address) : ILocalSiloDetails + { + public string Name => "test"; + public string ClusterId => "test"; + public string DnsHostName => "localhost"; + public SiloAddress SiloAddress => address; + public SiloAddress GatewayAddress => address; + } + + private sealed class TestMembershipService(ClusterMembershipSnapshot snapshot) : IClusterMembershipService + { + public ClusterMembershipSnapshot CurrentSnapshot => snapshot; + + public IAsyncEnumerable MembershipUpdates => Empty(); + + public ValueTask Refresh(MembershipVersion minimumVersion = default, CancellationToken cancellationToken = default) + => ValueTask.CompletedTask; + + public Task TryKill(SiloAddress siloAddress) => Task.FromResult(false); + + private static async IAsyncEnumerable Empty() + { + await Task.CompletedTask; + yield break; + } + + } + + private sealed class ManualTimeProvider(DateTimeOffset utcNow) : TimeProvider + { + public override DateTimeOffset GetUtcNow() => utcNow; + + public void Advance(TimeSpan amount) => utcNow += amount; + } +} diff --git a/test/Extensions/Orleans.AWS.Tests/Reminder/DynamoDBRemindersTableTests.cs b/test/Extensions/Orleans.AWS.Tests/Reminder/DynamoDBRemindersTableTests.cs index 01e9e2b63e3..e0aa3cf2b7d 100644 --- a/test/Extensions/Orleans.AWS.Tests/Reminder/DynamoDBRemindersTableTests.cs +++ b/test/Extensions/Orleans.AWS.Tests/Reminder/DynamoDBRemindersTableTests.cs @@ -6,6 +6,7 @@ using Orleans.Reminders.DynamoDB; using Orleans.Testing.Reminders; using Orleans.TestingHost; +using Orleans.TestingHost.Utils; using TestExtensions; using UnitTests; using UnitTests.RemindersTest; @@ -126,5 +127,488 @@ public async Task RemindersTable_AWS_ReminderSimple() { await ReminderSimple(); } + + [Fact] + public async Task DynamoDBReminderTable_Init_CreatesExpectedCurrentSchema() + { + if (!AWSTestConstants.IsDynamoDbAvailable) + throw Xunit.Sdk.SkipException.ForSkip("Unable to connect to AWS DynamoDB simulator"); + + const string ServiceId = "phase-1-schema-service"; + var tableName = $"OrleansReminders-{Guid.NewGuid():N}"; + var reminderTable = CreateIsolatedReminderTable(tableName, ServiceId); + using var client = CreateDynamoDBClient(); + + try + { + await reminderTable.Init().WaitAsync(TestContext.Current.CancellationToken); + + var response = await client.DescribeTableAsync( + new Amazon.DynamoDBv2.Model.DescribeTableRequest { TableName = tableName }, + TestContext.Current.CancellationToken); + var description = response.Table; + + Assert.Collection( + description.KeySchema, + key => + { + Assert.Equal("ReminderId", key.AttributeName); + Assert.Equal(Amazon.DynamoDBv2.KeyType.HASH, key.KeyType); + }, + key => + { + Assert.Equal("GrainHash", key.AttributeName); + Assert.Equal(Amazon.DynamoDBv2.KeyType.RANGE, key.KeyType); + }); + + Assert.Equal( + [ + ("GrainHash", "N"), + ("GrainReference", "S"), + ("ReminderId", "S"), + ("ServiceId", "S"), + ], + description.AttributeDefinitions + .OrderBy(attribute => attribute.AttributeName, StringComparer.Ordinal) + .Select(attribute => (attribute.AttributeName, attribute.AttributeType.Value)) + .ToArray()); + + var indexes = description.GlobalSecondaryIndexes + .OrderBy(index => index.IndexName, StringComparer.Ordinal) + .ToArray(); + Assert.Collection( + indexes, + index => + { + Assert.Equal("ServiceIdGrainReferenceIndex", index.IndexName); + Assert.Collection( + index.KeySchema, + key => + { + Assert.Equal("ServiceId", key.AttributeName); + Assert.Equal(Amazon.DynamoDBv2.KeyType.HASH, key.KeyType); + }, + key => + { + Assert.Equal("GrainReference", key.AttributeName); + Assert.Equal(Amazon.DynamoDBv2.KeyType.RANGE, key.KeyType); + }); + }, + index => + { + Assert.Equal("ServiceIdIndex", index.IndexName); + Assert.Collection( + index.KeySchema, + key => + { + Assert.Equal("ServiceId", key.AttributeName); + Assert.Equal(Amazon.DynamoDBv2.KeyType.HASH, key.KeyType); + }, + key => + { + Assert.Equal("GrainHash", key.AttributeName); + Assert.Equal(Amazon.DynamoDBv2.KeyType.RANGE, key.KeyType); + }); + }); + } + finally + { + await DeleteTableIfExistsAsync(client, tableName); + } + } + + [Fact] + public async Task DynamoDBReminderTable_UpsertRow_EncodesV1PrimaryKeyExactly() + { + if (!AWSTestConstants.IsDynamoDbAvailable) + throw Xunit.Sdk.SkipException.ForSkip("Unable to connect to AWS DynamoDB simulator"); + + const string ServiceId = "phase-1-key-service"; + const string ReminderName = "foo/bar\\#b_a_z?"; + var tableName = $"OrleansReminders-{Guid.NewGuid():N}"; + var grainId = Orleans.Runtime.GrainId.Create("phase-1", "deterministic-key"); + var reminderTable = CreateIsolatedReminderTable(tableName, ServiceId); + using var client = CreateDynamoDBClient(); + var initialized = false; + + try + { + await reminderTable.Init().WaitAsync(TestContext.Current.CancellationToken); + initialized = true; + + var entry = new Orleans.ReminderEntry + { + GrainId = grainId, + ReminderName = ReminderName, + StartAt = new DateTime(2026, 8, 28, 12, 34, 56, DateTimeKind.Utc), + Period = TimeSpan.FromMinutes(17), + }; + var etag = await reminderTable.UpsertRow(entry).WaitAsync(TestContext.Current.CancellationToken); + Assert.False(string.IsNullOrEmpty(etag)); + + var expectedReminderId = $"{ServiceId}_{grainId}_{ReminderName}"; + var expectedHash = grainId.GetUniformHashCode(); + var response = await client.GetItemAsync( + new Amazon.DynamoDBv2.Model.GetItemRequest + { + TableName = tableName, + ConsistentRead = true, + Key = new Dictionary + { + ["ReminderId"] = new(expectedReminderId), + ["GrainHash"] = new() { N = expectedHash.ToString(System.Globalization.CultureInfo.InvariantCulture) }, + }, + }, + TestContext.Current.CancellationToken); + + Assert.NotNull(response.Item); + Assert.Equal(8, response.Item.Count); + Assert.Equal(expectedReminderId, response.Item["ReminderId"].S); + Assert.Equal(expectedHash.ToString(System.Globalization.CultureInfo.InvariantCulture), response.Item["GrainHash"].N); + Assert.Equal(ServiceId, response.Item["ServiceId"].S); + Assert.Equal(grainId.ToString(), response.Item["GrainReference"].S); + Assert.Equal(ReminderName, response.Item["ReminderName"].S); + Assert.Equal(etag, response.Item["ETag"].N); + } + finally + { + try + { + if (initialized) + { + using var cleanupCancellation = new CancellationTokenSource(TimeSpan.FromSeconds(30)); + await reminderTable.TestOnlyClearTable().WaitAsync(cleanupCancellation.Token); + } + } + finally + { + await DeleteTableIfExistsAsync(client, tableName); + } + } + } + + [Fact] + public async Task DynamoDBReminderTable_PointRead_IsImmediatelyConsistent_AfterWriteAndDelete() + { + if (!AWSTestConstants.IsDynamoDbAvailable) + throw Xunit.Sdk.SkipException.ForSkip("Unable to connect to AWS DynamoDB simulator"); + + const string ServiceId = "phase-2-consistency-service"; + const string ReminderName = "immediate-point-read"; + var tableName = $"OrleansReminders-{Guid.NewGuid():N}"; + var grainId = Orleans.Runtime.GrainId.Create("phase-2", "point-consistency"); + var reminderTable = CreateIsolatedReminderTable(tableName, ServiceId); + using var client = CreateDynamoDBClient(); + var initializedTables = new List(); + + try + { + await reminderTable.Init().WaitAsync(TestContext.Current.CancellationToken); + initializedTables.Add(reminderTable); + + var entry = new Orleans.ReminderEntry + { + GrainId = grainId, + ReminderName = ReminderName, + StartAt = new DateTime(2026, 8, 28, 15, 16, 17, DateTimeKind.Utc), + Period = TimeSpan.FromMinutes(23), + }; + var etag = await reminderTable.UpsertRow(entry).WaitAsync(TestContext.Current.CancellationToken); + Assert.False(string.IsNullOrEmpty(etag)); + + var afterWrite = await reminderTable.ReadRow(grainId, ReminderName).WaitAsync(TestContext.Current.CancellationToken); + AssertReminder(entry, etag, Assert.IsType(afterWrite)); + + var removed = await reminderTable.RemoveRow(grainId, ReminderName, etag!).WaitAsync(TestContext.Current.CancellationToken); + Assert.True(removed); + + var afterDelete = await reminderTable.ReadRow(grainId, ReminderName).WaitAsync(TestContext.Current.CancellationToken); + Assert.Null(afterDelete); + } + finally + { + await ClearServicesAndDeleteTableAsync(client, tableName, initializedTables); + } + } + + [Fact] + public async Task DynamoDBReminderTable_TwoServicesSharingTable_AreFullyIsolated() + { + if (!AWSTestConstants.IsDynamoDbAvailable) + throw Xunit.Sdk.SkipException.ForSkip("Unable to connect to AWS DynamoDB simulator"); + + const string ServiceA = "phase-2-service-a"; + const string ServiceB = "phase-2-service-b"; + const string ReminderName = "shared-reminder"; + var tableName = $"OrleansReminders-{Guid.NewGuid():N}"; + var grainId = Orleans.Runtime.GrainId.Create("phase-2", "shared-grain"); + var serviceA = CreateIsolatedReminderTable(tableName, ServiceA); + var serviceB = CreateIsolatedReminderTable(tableName, ServiceB); + using var client = CreateDynamoDBClient(); + var initializedTables = new List(); + + try + { + await serviceA.Init().WaitAsync(TestContext.Current.CancellationToken); + initializedTables.Add(serviceA); + await serviceB.Init().WaitAsync(TestContext.Current.CancellationToken); + initializedTables.Add(serviceB); + + var entryA = new Orleans.ReminderEntry + { + GrainId = grainId, + ReminderName = ReminderName, + StartAt = new DateTime(2026, 8, 28, 1, 2, 3, DateTimeKind.Utc), + Period = TimeSpan.FromMinutes(11), + }; + var entryB = new Orleans.ReminderEntry + { + GrainId = grainId, + ReminderName = ReminderName, + StartAt = new DateTime(2026, 8, 29, 4, 5, 6, DateTimeKind.Utc), + Period = TimeSpan.FromMinutes(37), + }; + var etagA = await serviceA.UpsertRow(entryA).WaitAsync(TestContext.Current.CancellationToken); + var etagB = await serviceB.UpsertRow(entryB).WaitAsync(TestContext.Current.CancellationToken); + Assert.False(string.IsNullOrEmpty(etagA)); + Assert.False(string.IsNullOrEmpty(etagB)); + + var pointA = await serviceA.ReadRow(grainId, ReminderName).WaitAsync(TestContext.Current.CancellationToken); + var pointB = await serviceB.ReadRow(grainId, ReminderName).WaitAsync(TestContext.Current.CancellationToken); + AssertReminder(entryA, etagA, Assert.IsType(pointA)); + AssertReminder(entryB, etagB, Assert.IsType(pointB)); + + var expectedA = new (Orleans.ReminderEntry Entry, string? ETag)[] { (entryA, etagA) }; + var expectedB = new (Orleans.ReminderEntry Entry, string? ETag)[] { (entryB, etagB) }; + AssertReminderRows( + expectedA, + await ReadRowsUntilExactlyAsync( + () => serviceA.ReadRows(grainId), + expectedA, + TestContext.Current.CancellationToken)); + AssertReminderRows( + expectedB, + await ReadRowsUntilExactlyAsync( + () => serviceB.ReadRows(grainId), + expectedB, + TestContext.Current.CancellationToken)); + AssertReminderRows( + expectedA, + await ReadRowsUntilExactlyAsync( + () => serviceA.ReadRows(0, 0), + expectedA, + TestContext.Current.CancellationToken)); + AssertReminderRows( + expectedB, + await ReadRowsUntilExactlyAsync( + () => serviceB.ReadRows(0, 0), + expectedB, + TestContext.Current.CancellationToken)); + + await serviceA.TestOnlyClearTable().WaitAsync(TestContext.Current.CancellationToken); + + var clearedPointA = await serviceA.ReadRow(grainId, ReminderName).WaitAsync(TestContext.Current.CancellationToken); + var retainedPointB = await serviceB.ReadRow(grainId, ReminderName).WaitAsync(TestContext.Current.CancellationToken); + Assert.Null(clearedPointA); + AssertReminder(entryB, etagB, Assert.IsType(retainedPointB)); + + var noReminders = Array.Empty<(Orleans.ReminderEntry Entry, string? ETag)>(); + AssertReminderRows( + noReminders, + await ReadRowsUntilExactlyAsync( + () => serviceA.ReadRows(grainId), + noReminders, + TestContext.Current.CancellationToken)); + AssertReminderRows( + expectedB, + await ReadRowsUntilExactlyAsync( + () => serviceB.ReadRows(grainId), + expectedB, + TestContext.Current.CancellationToken)); + AssertReminderRows( + noReminders, + await ReadRowsUntilExactlyAsync( + () => serviceA.ReadRows(0, 0), + noReminders, + TestContext.Current.CancellationToken)); + AssertReminderRows( + expectedB, + await ReadRowsUntilExactlyAsync( + () => serviceB.ReadRows(0, 0), + expectedB, + TestContext.Current.CancellationToken)); + } + finally + { + await ClearServicesAndDeleteTableAsync(client, tableName, initializedTables); + } + } + + private DynamoDBReminderTable CreateIsolatedReminderTable(string tableName, string serviceId) + { + var storageOptions = new DynamoDBReminderStorageOptions + { + Service = AWSTestConstants.DynamoDbService, + AccessKey = AWSTestConstants.DynamoDbAccessKey, + SecretKey = AWSTestConstants.DynamoDbSecretKey, + TableName = tableName, + CreateIfNotExists = true, + UpdateIfExists = false, + UseProvisionedThroughput = false, + }; + + return new DynamoDBReminderTable( + loggerFactory, + Options.Create(new ClusterOptions { ClusterId = serviceId, ServiceId = serviceId }), + Options.Create(storageOptions)); + } + + private static async Task ReadRowsUntilExactlyAsync( + Func> read, + IReadOnlyList<(Orleans.ReminderEntry Entry, string? ETag)> expected, + CancellationToken cancellationToken) + { + Orleans.ReminderTableData? observed = null; + await TestingUtils.WaitUntilAsync( + async (lastTry, attemptCancellation) => + { + observed = await read().WaitAsync(attemptCancellation); + var matches = ReminderRowsMatch(expected, observed); + if (lastTry && !matches) + { + AssertReminderRows(expected, observed); + } + + return matches; + }, + TimeSpan.FromSeconds(10), + TimeSpan.FromMilliseconds(100), + cancellationToken); + return Assert.IsType(observed); + } + + private static bool ReminderRowsMatch( + IReadOnlyList<(Orleans.ReminderEntry Entry, string? ETag)> expected, + Orleans.ReminderTableData actual) + => actual.Reminders.Count == expected.Count + && expected.All(item => actual.Reminders.Any( + candidate => ReminderMatches(item.Entry, item.ETag, candidate))); + + private static bool ReminderMatches( + Orleans.ReminderEntry expected, + string? expectedETag, + Orleans.ReminderEntry actual) + => actual.GrainId.Equals(expected.GrainId) + && string.Equals(actual.ReminderName, expected.ReminderName, StringComparison.Ordinal) + && actual.StartAt.Ticks == expected.StartAt.Ticks + && actual.Period == expected.Period + && string.Equals(actual.ETag, expectedETag, StringComparison.Ordinal); + + private static void AssertReminderRows( + IReadOnlyList<(Orleans.ReminderEntry Entry, string? ETag)> expected, + Orleans.ReminderTableData actual) + { + Assert.Equal(expected.Count, actual.Reminders.Count); + foreach (var item in expected) + { + var actualEntry = Assert.Single( + actual.Reminders, + candidate => candidate.GrainId.Equals(item.Entry.GrainId) + && string.Equals(candidate.ReminderName, item.Entry.ReminderName, StringComparison.Ordinal)); + AssertReminder(item.Entry, item.ETag, actualEntry); + } + } + + private static void AssertReminder( + Orleans.ReminderEntry expected, + string? expectedETag, + Orleans.ReminderEntry actual) + { + Assert.Equal(expected.GrainId, actual.GrainId); + Assert.Equal(expected.ReminderName, actual.ReminderName); + Assert.Equal(expected.StartAt.Ticks, actual.StartAt.Ticks); + Assert.Equal(expected.Period, actual.Period); + Assert.Equal(expectedETag, actual.ETag); + } + + private static async Task ClearServicesAndDeleteTableAsync( + Amazon.DynamoDBv2.AmazonDynamoDBClient client, + string tableName, + IReadOnlyList initializedTables) + { + List? failures = null; + foreach (var table in initializedTables) + { + try + { + using var cleanupCancellation = new CancellationTokenSource(TimeSpan.FromSeconds(30)); + await table.TestOnlyClearTable().WaitAsync(cleanupCancellation.Token); + } + catch (Exception exception) + { + (failures ??= []).Add(exception); + } + } + + try + { + await DeleteTableIfExistsAsync(client, tableName); + } + catch (Exception exception) + { + (failures ??= []).Add(exception); + } + + if (failures is { Count: 1 }) + { + throw failures[0]; + } + + if (failures is { Count: > 1 }) + { + throw new AggregateException(failures); + } + } + + private static Amazon.DynamoDBv2.AmazonDynamoDBClient CreateDynamoDBClient() + { + var service = AWSTestConstants.DynamoDbService; + if (service.StartsWith("http://", StringComparison.OrdinalIgnoreCase) + || service.StartsWith("https://", StringComparison.OrdinalIgnoreCase)) + { + return new Amazon.DynamoDBv2.AmazonDynamoDBClient( + new Amazon.Runtime.BasicAWSCredentials("dummy", "dummyKey"), + new Amazon.DynamoDBv2.AmazonDynamoDBConfig { ServiceURL = service }); + } + + var config = new Amazon.DynamoDBv2.AmazonDynamoDBConfig + { + RegionEndpoint = Amazon.RegionEndpoint.GetBySystemName(service), + }; + return string.IsNullOrEmpty(AWSTestConstants.DynamoDbAccessKey) + || string.IsNullOrEmpty(AWSTestConstants.DynamoDbSecretKey) + ? new Amazon.DynamoDBv2.AmazonDynamoDBClient(config) + : new Amazon.DynamoDBv2.AmazonDynamoDBClient( + new Amazon.Runtime.BasicAWSCredentials( + AWSTestConstants.DynamoDbAccessKey, + AWSTestConstants.DynamoDbSecretKey), + config); + } + + private static async Task DeleteTableIfExistsAsync( + Amazon.DynamoDBv2.AmazonDynamoDBClient client, + string tableName) + { + using var cleanupCancellation = new CancellationTokenSource(TimeSpan.FromSeconds(30)); + try + { + await client.DeleteTableAsync( + new Amazon.DynamoDBv2.Model.DeleteTableRequest { TableName = tableName }, + cleanupCancellation.Token); + } + catch (Amazon.DynamoDBv2.Model.ResourceNotFoundException) + { + } + } } } diff --git a/test/Orleans.Reminders.Tests/TimerTests/ControllableReminderTable.cs b/test/Orleans.Reminders.Tests/TimerTests/ControllableReminderTable.cs index 2923d621773..d921e346526 100644 --- a/test/Orleans.Reminders.Tests/TimerTests/ControllableReminderTable.cs +++ b/test/Orleans.Reminders.Tests/TimerTests/ControllableReminderTable.cs @@ -8,6 +8,10 @@ internal sealed class ControllableReminderTable( InMemoryReminderTable inner, ReminderTableReadController readController) : IReminderTable { + private int pointReadCount; + + public int PointReadCount => Volatile.Read(ref pointReadCount); + public Task StartAsync(CancellationToken cancellationToken = default) => ((IReminderTable)inner).StartAsync(cancellationToken); public Task ReadRows(GrainId grainId) => inner.ReadRows(grainId); @@ -15,11 +19,18 @@ internal sealed class ControllableReminderTable( public async Task ReadRows(uint begin, uint end) { var result = await inner.ReadRows(begin, end); + result = readController.TransformRangeRead(begin, end, result); await readController.OnRangeReadAsync(begin, end); return result; } - public Task ReadRow(GrainId grainId, string reminderName) => inner.ReadRow(grainId, reminderName); + public async Task ReadRow(GrainId grainId, string reminderName) + { + Interlocked.Increment(ref pointReadCount); + var result = await inner.ReadRow(grainId, reminderName); + await readController.OnPointReadAsync(grainId, reminderName); + return result; + } public Task UpsertRow(ReminderEntry entry) => inner.UpsertRow(entry); @@ -34,6 +45,15 @@ internal sealed class ReminderTableReadController { private readonly object _lock = new(); private readonly List _gates = []; + private readonly List _pointReadGates = []; + private readonly List<(GrainId GrainId, string ReminderName)> _omissions = []; + public void OmitFromNextRangeRead(GrainId grainId, string reminderName) + { + lock (_lock) + { + _omissions.Add((grainId, reminderName)); + } + } public ReminderTableReadGate BlockNextRangeRead(GrainId grainId, CancellationToken cancellationToken) { @@ -46,6 +66,20 @@ public ReminderTableReadGate BlockNextRangeRead(GrainId grainId, CancellationTok return gate; } + public ReminderPointReadGate BlockNextPointRead( + GrainId grainId, + string reminderName, + CancellationToken cancellationToken) + { + var gate = new ReminderPointReadGate(this, grainId, reminderName, cancellationToken); + lock (_lock) + { + _pointReadGates.Add(gate); + } + + return gate; + } + internal async Task OnRangeReadAsync(uint begin, uint end) { ReminderTableReadGate? gate = null; @@ -59,6 +93,7 @@ internal async Task OnRangeReadAsync(uint begin, uint end) _gates.RemoveAt(i); break; } + } } @@ -69,12 +104,74 @@ internal async Task OnRangeReadAsync(uint begin, uint end) } } + internal ReminderTableData TransformRangeRead( + uint begin, + uint end, + ReminderTableData result) + { + lock (_lock) + { + for (var i = 0; i < _omissions.Count; i++) + { + var omission = _omissions[i]; + if (!Matches(omission.GrainId.GetUniformHashCode(), begin, end)) + { + continue; + } + + _omissions.RemoveAt(i); + return new(result.Reminders.Where(entry => + entry.GrainId != omission.GrainId + || !string.Equals(entry.ReminderName, omission.ReminderName, StringComparison.Ordinal))); + } + } + + return result; + } + + internal async Task OnPointReadAsync(GrainId grainId, string reminderName) + { + ReminderPointReadGate? gate = null; + lock (_lock) + { + for (var i = 0; i < _pointReadGates.Count; i++) + { + if (_pointReadGates[i].Matches(grainId, reminderName)) + { + gate = _pointReadGates[i]; + _pointReadGates.RemoveAt(i); + break; + } + } + } + + if (gate is not null) + { + gate.MarkBlocked(); + await gate.WaitForReleaseAsync(); + } + } + + private static bool Matches(uint grainHash, uint begin, uint end) + => begin < end + ? grainHash > begin && grainHash <= end + : grainHash > begin || grainHash <= end; + internal void Remove(ReminderTableReadGate gate) { lock (_lock) { _gates.Remove(gate); } + + } + + internal void Remove(ReminderPointReadGate gate) + { + lock (_lock) + { + _pointReadGates.Remove(gate); + } } } @@ -112,3 +209,38 @@ public ValueTask DisposeAsync() return ValueTask.CompletedTask; } } + +internal sealed class ReminderPointReadGate( + ReminderTableReadController owner, + GrainId grainId, + string reminderName, + CancellationToken cancellationToken) : IAsyncDisposable +{ + private readonly TaskCompletionSource _blocked = new(TaskCreationOptions.RunContinuationsAsynchronously); + private readonly TaskCompletionSource _release = new(TaskCreationOptions.RunContinuationsAsynchronously); + private int _disposed; + + internal bool Matches(GrainId candidateGrainId, string candidateReminderName) + => grainId == candidateGrainId + && string.Equals(reminderName, candidateReminderName, StringComparison.Ordinal); + + internal void MarkBlocked() => _blocked.TrySetResult(); + + internal Task WaitForReleaseAsync() => _release.Task.WaitAsync(cancellationToken); + + public Task WaitUntilBlockedAsync(CancellationToken cancellationToken) + => _blocked.Task.WaitAsync(cancellationToken); + + public void Release() => _release.TrySetResult(); + + public ValueTask DisposeAsync() + { + if (Interlocked.Exchange(ref _disposed, 1) == 0) + { + owner.Remove(this); + Release(); + } + + return ValueTask.CompletedTask; + } +} diff --git a/test/Orleans.Reminders.Tests/TimerTests/ReminderTests_TableGrain.cs b/test/Orleans.Reminders.Tests/TimerTests/ReminderTests_TableGrain.cs index 4f4a883ef34..0d3ef68ff37 100644 --- a/test/Orleans.Reminders.Tests/TimerTests/ReminderTests_TableGrain.cs +++ b/test/Orleans.Reminders.Tests/TimerTests/ReminderTests_TableGrain.cs @@ -462,6 +462,92 @@ public async Task Rem_Grain_StaleRefreshCannotRestoreUnregisteredReminder() followingRead.Release(); } + [Fact] + public async Task Rem_Grain_MissingDiscoveryCandidateRequiresStrongPointAbsenceBeforeRemoval() + { + const string reminderName = "missing_discovery_candidate"; + var grain = GrainFactory.GetGrain(Guid.NewGuid()); + var grainId = grain.GetGrainId(); + var silo = Assert.Single(HostedCluster.Silos); + var reminderService = silo.ServiceProvider.GetRequiredService(); + var reminderTable = silo.ServiceProvider.GetRequiredService(); + using var cancellation = CancellationTokenSource.CreateLinkedTokenSource(TestContext.Current.CancellationToken); + cancellation.CancelAfter(TestConstants.InitTimeout); + + var activated = observer.WaitForActiveReminderCountAsync(grainId, 1, cancellation.Token, reminderName); + await grain.StartReminder(reminderName, ReminderLoadingWindow, TimeSpan.FromMinutes(2)).WaitAsync(cancellation.Token); + await activated; + + _readController.OmitFromNextRangeRead(grainId, reminderName); + await reminderService.TestOnlyRefresh().WaitAsync(cancellation.Token); + Assert.Equal(1, observer.GetActiveReminderCount(grainId, reminderName)); + Assert.NotNull(await grain.GetReminderObject(reminderName).WaitAsync(cancellation.Token)); + + var persisted = Assert.IsType(await reminderTable.ReadRow(grainId, reminderName)); + Assert.True(await reminderTable.RemoveRow(grainId, reminderName, persisted.ETag!)); + var quiesced = observer.WaitForReminderQuiescenceAsync(grainId, reminderName, cancellation.Token); + _readController.OmitFromNextRangeRead(grainId, reminderName); + await reminderService.TestOnlyRefresh().WaitAsync(cancellation.Token); + await quiesced; + + Assert.Equal(0, observer.GetActiveReminderCount(grainId, reminderName)); + Assert.Null(await grain.GetReminderObject(reminderName).WaitAsync(cancellation.Token)); + } + + [Fact] + public async Task RegisterAndUnregisterReconcileThroughPointReadsBeforeReturning() + { + const string reminderName = "immediate_point_reconciliation"; + var grain = GrainFactory.GetGrain(Guid.NewGuid()); + var silo = Assert.Single(HostedCluster.Silos); + var reminderTable = silo.ServiceProvider.GetRequiredService(); + var initialPointReads = reminderTable.PointReadCount; + + await grain.StartReminder(reminderName, ReminderLoadingWindow, TimeSpan.FromMinutes(2)) + .WaitAsync(TestContext.Current.CancellationToken); + var afterRegistration = reminderTable.PointReadCount; + Assert.True(afterRegistration > initialPointReads); + + await grain.StopReminder(reminderName).WaitAsync(TestContext.Current.CancellationToken); + Assert.True(reminderTable.PointReadCount > afterRegistration); + } + + [Fact] + public async Task ConcurrentMutationPointReadsCannotRestoreAnOlderSchedule() + { + const string reminderName = "out_of_order_point_reads"; + var grainId = GrainId.Create("point-read-race", Guid.NewGuid().ToString("N")); + var silo = Assert.Single(HostedCluster.Silos); + var reminderService = silo.ServiceProvider.GetRequiredService(); + using var cancellation = CancellationTokenSource.CreateLinkedTokenSource(TestContext.Current.CancellationToken); + cancellation.CancelAfter(TestConstants.InitTimeout); + await using var olderRead = _readController.BlockNextPointRead(grainId, reminderName, cancellation.Token); + + var olderMutation = reminderService.TestOnlyRegisterOrUpdateReminder( + grainId, + reminderName, + TimeSpan.FromMinutes(1), + TimeSpan.FromMinutes(2)); + await olderRead.WaitUntilBlockedAsync(cancellation.Token); + + var newerReminder = await reminderService.TestOnlyRegisterOrUpdateReminder( + grainId, + reminderName, + TimeSpan.FromMinutes(3), + TimeSpan.FromMinutes(4)); + var afterNewerMutation = Assert.IsType( + await reminderService.TestOnlyGetLocalReminder(grainId, reminderName)); + Assert.Equal(TimeSpan.FromMinutes(4), afterNewerMutation.Period); + + olderRead.Release(); + await olderMutation.WaitAsync(cancellation.Token); + var afterOlderReadCompletes = Assert.IsType( + await reminderService.TestOnlyGetLocalReminder(grainId, reminderName)); + Assert.Equal(TimeSpan.FromMinutes(4), afterOlderReadCompletes.Period); + + await reminderService.TestOnlyUnregisterReminder(newerReminder).WaitAsync(cancellation.Token); + } + [Fact] public async Task Rem_Grain_StaleRefreshCannotReloadStorageOnlyScheduleAfterDistantUpdate() {