| | | 1 | | using MongoDB.Bson; |
| | | 2 | | using MongoDB.Driver; |
| | | 3 | | |
| | | 4 | | namespace AsyncResponse.Internal; |
| | | 5 | | |
| | | 6 | | /// <summary> |
| | | 7 | | /// Persisted cross-host collection-ownership ledger. The in-container |
| | | 8 | | /// <c>MongoNamespaceRegistry</c> fails same-container collisions at construction, but two HOSTS |
| | | 9 | | /// (or a directly constructed store) sharing a database cannot see each other's claims — and |
| | | 10 | | /// MongoDB has no catalog metadata to verify after the fact: a durable-flow store configured |
| | | 11 | | /// onto the channel's derived <c>{MessageCollection}_counters</c> collection happily writes flow |
| | | 12 | | /// documents there, and its TTL index then silently deletes the ack-sequence counter. Each |
| | | 13 | | /// store, at first use (EnsureCreated), atomically upserts one claim document per effective |
| | | 14 | | /// collection into the fixed <c>asyncresponse_ownership</c> collection; a claim already held by |
| | | 15 | | /// a DIFFERENT component (or the same component in a different role) fails startup with an |
| | | 16 | | /// actionable error naming both claimants — whichever process starts second, whatever the |
| | | 17 | | /// order. Restarts re-claim idempotently. Deployments that disable auto-creation own their |
| | | 18 | | /// provisioning and skip the ledger, like the rest of the first-use DDL. Renaming a collection |
| | | 19 | | /// in configuration leaves the old claim behind; the error text covers removing a stale claim |
| | | 20 | | /// document deliberately. Source-linked into the channel, transport, and durable-flow packages. |
| | | 21 | | /// </summary> |
| | | 22 | | internal static class MongoOwnershipLedger |
| | | 23 | | { |
| | | 24 | | /// <summary>Fixed ledger collection name; rejected as a configurable data collection by every store.</summary> |
| | | 25 | | public const string CollectionName = "asyncresponse_ownership"; |
| | | 26 | | |
| | | 27 | | public static async Task ClaimAsync( |
| | | 28 | | IMongoDatabase database, |
| | | 29 | | string componentName, |
| | | 30 | | IReadOnlyList<(string Collection, string Purpose)> claims, |
| | | 31 | | CancellationToken cancellationToken) |
| | | 32 | | { |
| | 155 | 33 | | var ledger = database.GetCollection<BsonDocument>(CollectionName); |
| | 619 | 34 | | foreach (var (collection, purpose) in claims) |
| | | 35 | | { |
| | | 36 | | // Atomic claim: the upsert inserts our ownership document only when no document |
| | | 37 | | // exists for the collection; a concurrent claimant loses the insert and reads the |
| | | 38 | | // winner's document. Documents lacking the component/purpose fields — or carrying |
| | | 39 | | // non-string values (a hand-repaired claim with component: null, a migration that |
| | | 40 | | // stored an enum) — are foreign writes: tolerated rather than guessed about, and the |
| | | 41 | | // BsonString pattern matches below keep the guard itself from throwing |
| | | 42 | | // InvalidCastException while evaluating them. |
| | 155 | 43 | | Task<BsonDocument?> UpsertClaimAsync() => ledger.FindOneAndUpdateAsync<BsonDocument?>( |
| | 155 | 44 | | new BsonDocument("_id", collection), |
| | 155 | 45 | | new BsonDocument("$setOnInsert", new BsonDocument |
| | 155 | 46 | | { |
| | 155 | 47 | | { "component", componentName }, |
| | 155 | 48 | | { "purpose", purpose } |
| | 155 | 49 | | }), |
| | 155 | 50 | | new FindOneAndUpdateOptions<BsonDocument, BsonDocument?> |
| | 155 | 51 | | { |
| | 155 | 52 | | IsUpsert = true, |
| | 155 | 53 | | ReturnDocument = ReturnDocument.Before |
| | 155 | 54 | | }, |
| | 155 | 55 | | cancellationToken); |
| | | 56 | | |
| | | 57 | | BsonDocument? existing; |
| | | 58 | | try |
| | | 59 | | { |
| | 155 | 60 | | existing = await UpsertClaimAsync().ConfigureAwait(false); |
| | 155 | 61 | | } |
| | 0 | 62 | | catch (MongoException ex) when (IsDuplicateKey(ex)) |
| | | 63 | | { |
| | | 64 | | // The upsert's no-match-then-insert is not atomic against a concurrent FIRST |
| | | 65 | | // claim on the same _id: the loser of that race gets E11000 instead of the |
| | | 66 | | // winner's document. One identical retry now matches the winner's document and |
| | | 67 | | // resolves through the ownership check below — idempotent success for the same |
| | | 68 | | // component, the actionable conflict error for a different one. |
| | 0 | 69 | | existing = await UpsertClaimAsync().ConfigureAwait(false); |
| | | 70 | | } |
| | | 71 | | |
| | 155 | 72 | | if (existing is not null |
| | 155 | 73 | | && existing.TryGetValue("component", out var owner) |
| | 155 | 74 | | && existing.TryGetValue("purpose", out var ownerPurpose) |
| | 155 | 75 | | && owner is BsonString ownerName |
| | 155 | 76 | | && ownerPurpose is BsonString ownerPurposeName |
| | 155 | 77 | | && !(ownerName.Value == componentName && ownerPurposeName.Value == purpose)) |
| | | 78 | | { |
| | 1 | 79 | | throw new InvalidOperationException( |
| | 1 | 80 | | $"MongoDB collection '{database.DatabaseNamespace.DatabaseName}.{collection}' is already claimed by |
| | 1 | 81 | | $"{ownerName.Value} ({ownerPurposeName.Value}) in the persisted ownership ledger " + |
| | 1 | 82 | | $"('{CollectionName}'), and this {componentName} configured it as {purpose}. Components sharing a da |
| | 1 | 83 | | "must use distinct collections — including derived ones such as the channel's '{MessageCollection}_c |
| | 1 | 84 | | "ack-sequence counter, whose documents another component's TTL index would silently delete. Rename o |
| | 1 | 85 | | "configured collection names, or delete the stale claim document if the other component was delibera |
| | 1 | 86 | | "reconfigured away from this collection."); |
| | | 87 | | } |
| | 154 | 88 | | } |
| | 154 | 89 | | } |
| | | 90 | | |
| | | 91 | | /// <summary> |
| | | 92 | | /// The server's duplicate-key rejection (the E11000 family), on either surface it reaches the |
| | | 93 | | /// driver through: findAndModify reports it as a command error, write commands as a |
| | | 94 | | /// categorized write error. The code set matches the driver's own |
| | | 95 | | /// <see cref="ServerErrorCategory.DuplicateKey"/> mapping. |
| | | 96 | | /// </summary> |
| | | 97 | | private static bool IsDuplicateKey(MongoException exception) |
| | 0 | 98 | | => exception is MongoWriteException { WriteError.Category: ServerErrorCategory.DuplicateKey } |
| | 0 | 99 | | || exception is MongoCommandException { Code: 11000 or 11001 or 12582 }; |
| | | 100 | | } |