From 2dcd2be7104113f9fb69721596d3b9ee37741d50 Mon Sep 17 00:00:00 2001 From: "Mars.P" Date: Mon, 19 Oct 2026 08:17:18 +0800 Subject: [PATCH] Let a re-authorization win over the repair sweeps The repair of stranded bindings, the connect-time carry-forward and the departed-member retirement each judge a credential dead and act on that verdict a moment later. An in-place re-authorization that lands in between revives the credential, yet the move or the retirement went on: the bindings left a credential that was healthy again, or its new token was blanked and the person had been told the re-authorization worked. Both sweeps now take the credential's refresh lock, the one every rewrite of its token takes, without waiting. A credential somebody is writing is skipped and met again on a later pass; otherwise the verdict is taken again under the lock, by the same query that picked the candidate, so a credential revived since the page was read moves nothing. The successor is the one that verdict names. The refresh lock is taken by the writers of the token only, and the hourly health probe writes a refused credential back to Active, when its provider accepts it again, without it. So the repair also holds the dead credential's row (FOR NO KEY UPDATE, in a transaction of its own per credential) and judges it under that hold: a verdict recorded since the page was read is seen, and one that arrives later waits for the commit and finds the bindings already moved. A connect cannot hold the lock to the end of its unit of work without leaking it on a rollback, so it holds the dead credential's row for the transaction instead and drops any whose status or payload is no longer what it read. A re-authorization that arrives after waits for the commit and finds the bindings already moved. Skipping a busy credential is not a failure: it never counts toward the sweep's back-off, which the tests pin by skipping more passes than the back-off tolerates before the lock is released. No schema, API or lock-key change: the advisory key is the one the refresher and the previous build already use, so pods of either build exclude each other. --- .../CredentialRevocationService.cs | 32 +- .../CredentialSuccessionService.cs | 128 ++++- .../ICredentialRevocationService.cs | 4 +- .../ICredentialSuccessionService.cs | 9 +- .../Services/OAuth/CredentialRefreshLock.cs | 35 +- .../Services/OAuth/ICrossProcessTryLock.cs | 4 +- ...irDisconnectedCredentialBindingsCommand.cs | 3 +- ...RetireDepartedMembersCredentialsCommand.cs | 3 +- .../CredentialSweepLockE2ETests.cs | 272 ++++++++++ ...edentialReauthorizationFlowTests.Sweeps.cs | 488 ++++++++++++++++++ .../CredentialReauthorizationFlowTests.cs | 12 +- .../OAuth/RepairSweepFairnessFlowTests.cs | 12 + .../OAuth/RetireSweepFairnessFlowTests.cs | 10 + .../ScopedTransactionInventoryTests.cs | 1 + .../OAuth/CredentialRefreshLockTests.cs | 115 ++++- 15 files changed, 1099 insertions(+), 29 deletions(-) create mode 100644 backend/tests/CodeSpace.E2ETests/Credentials/CredentialSweepLockE2ETests.cs create mode 100644 backend/tests/CodeSpace.IntegrationTests/Credentials/CredentialReauthorizationFlowTests.Sweeps.cs diff --git a/backend/src/CodeSpace.Core/Services/Credentials/CredentialRevocationService.cs b/backend/src/CodeSpace.Core/Services/Credentials/CredentialRevocationService.cs index 640ccb24d..05dc113bd 100644 --- a/backend/src/CodeSpace.Core/Services/Credentials/CredentialRevocationService.cs +++ b/backend/src/CodeSpace.Core/Services/Credentials/CredentialRevocationService.cs @@ -1,6 +1,7 @@ using CodeSpace.Core.DependencyInjection; using CodeSpace.Core.Persistence.Db; using CodeSpace.Core.Persistence.Entities; +using CodeSpace.Core.Services.OAuth; using CodeSpace.Messages.Enums; using Microsoft.EntityFrameworkCore; using Microsoft.Extensions.Logging; @@ -22,12 +23,14 @@ public sealed class CredentialRevocationService : ICredentialRevocationService, private readonly CodeSpaceDbContext _db; private readonly ISweepFailureBackoff _backoff; + private readonly ICrossProcessTryLock _tryLock; private readonly ILogger _logger; - public CredentialRevocationService(CodeSpaceDbContext db, ISweepFailureBackoff backoff, ILogger logger) + public CredentialRevocationService(CodeSpaceDbContext db, ISweepFailureBackoff backoff, ICrossProcessTryLock tryLock, ILogger logger) { _db = db; _backoff = backoff; + _tryLock = tryLock; _logger = logger; } @@ -80,10 +83,33 @@ private async Task> LoadDepartedAsync(Guid? after, int } /// - /// Re-read under the candidate rule and retired on its own save, so one failure undoes no other. A person who rejoined - /// since the candidate list was read keeps the token. + /// Retire one candidate under the credential's refresh lock. A token somebody is writing — a refresh, or a re-authorization + /// by a person who has just rejoined — is skipped, not waited for, and met again on the next tick: retiring it under that write + /// would blank the token the write lands, and the person would be told their re-authorization worked. /// private async Task RetireDepartedOneAsync(DepartedCredential candidate, CancellationToken cancellationToken) + { + await using var held = await CredentialRefreshLock.TryAcquireAsync(_tryLock, candidate.Id, cancellationToken).ConfigureAwait(false); + + if (held is null) return SkipBusy(candidate); + + return await RetireHeldAsync(candidate, cancellationToken).ConfigureAwait(false); + } + + private bool SkipBusy(DepartedCredential candidate) + { + _logger.LogInformation("{Sweep}: credential {CredentialId} is being written (a refresh or a re-authorization holds its lock); it stays a candidate for the next pass", SweepName, candidate.Id); + + return false; + } + + /// + /// Re-read under the candidate rule — now under the lock, so no write of the token can land between the verdict and the + /// retirement — and retired on its own save, so one failure undoes no other. A person who rejoined since the candidate list + /// was read keeps the token, and so does a re-authorization they completed before the lock was taken. The verdict is the + /// person's standing and the credential's ownership, which no write of its health changes, so the lock is the whole exclusion here. + /// + private async Task RetireHeldAsync(DepartedCredential candidate, CancellationToken cancellationToken) { try { diff --git a/backend/src/CodeSpace.Core/Services/Credentials/CredentialSuccessionService.cs b/backend/src/CodeSpace.Core/Services/Credentials/CredentialSuccessionService.cs index 464958a1a..362b6d039 100644 --- a/backend/src/CodeSpace.Core/Services/Credentials/CredentialSuccessionService.cs +++ b/backend/src/CodeSpace.Core/Services/Credentials/CredentialSuccessionService.cs @@ -2,6 +2,7 @@ using CodeSpace.Core.Middlewares.Transactional; using CodeSpace.Core.Persistence.Db; using CodeSpace.Core.Persistence.Entities; +using CodeSpace.Core.Services.OAuth; using CodeSpace.Core.Services.Webhooks.Registration; using CodeSpace.Messages.Enums; using Microsoft.EntityFrameworkCore; @@ -22,6 +23,16 @@ namespace CodeSpace.Core.Services.Credentials; /// one other connection implies) or a person relinks them /// (), each a deliberate act rather than a side effect of connecting. The repair sweep /// stays the fallback for what none of those doors reached, and walks its backlog in bounded pages (). +/// +/// Both judge a credential dead and act on the verdict a moment later, and an in-place re-authorization +/// () can land in between — after which the credential is healthy and what is bound to it +/// must stay. The sweep takes the credential's refresh lock () and judges it again under it, +/// skipping a credential somebody is writing and meeting it on a later pass, so the re-authorization is either before the +/// verdict or after the move. That lock is taken by the writers of the token only, and the credential's row has other writers — +/// the hourly health probe writes a refused credential back to Active when its provider accepts it again — so the sweep also holds the +/// row for the move (). A connect cannot hold the lock until its unit of work commits without leaking it +/// on a rollback, so it holds the dead credential's row for the transaction instead (): that excludes +/// every writer of the row, ended by the commit or the rollback whichever comes. /// public sealed partial class CredentialSuccessionService : ICredentialSuccessionService, IStrandedCredentialCensus, IScopedDependency { @@ -34,14 +45,16 @@ public sealed partial class CredentialSuccessionService : ICredentialSuccessionS private readonly IConnectionWebhookRegistrationDispatcher _hookDispatcher; private readonly IPostCommitActions _postCommit; private readonly ISweepFailureBackoff _backoff; + private readonly ICrossProcessTryLock _tryLock; private readonly ILogger _logger; - public CredentialSuccessionService(CodeSpaceDbContext db, IConnectionWebhookRegistrationDispatcher hookDispatcher, IPostCommitActions postCommit, ISweepFailureBackoff backoff, ILogger logger) + public CredentialSuccessionService(CodeSpaceDbContext db, IConnectionWebhookRegistrationDispatcher hookDispatcher, IPostCommitActions postCommit, ISweepFailureBackoff backoff, ICrossProcessTryLock tryLock, ILogger logger) { _db = db; _hookDispatcher = hookDispatcher; _postCommit = postCommit; _backoff = backoff; + _tryLock = tryLock; _logger = logger; } @@ -67,7 +80,8 @@ public async Task RepairStrandedAsync(int batchSize, CancellationToken canc /// The owner's dead credentials on the successor's instance and team — revoked, or refused by their provider. Filtered on /// the tracked entities, not in SQL: a token link revokes the credential it replaces in the same unit of work, so that row /// is Revoked in memory before it is Revoked in the database. No date is asked of a refused one here: the successor is - /// being created now, after whatever the owner's credentials last did. + /// being created now, after whatever the owner's credentials last did. Held as read before it is moved + /// (), because a re-authorization can have revived one since. /// private async Task> LoadDeadPredecessorIdsAsync(Credential successor, Guid ownerId, CancellationToken cancellationToken) { @@ -75,7 +89,42 @@ private async Task> LoadDeadPredecessorIdsAsync(Credential successor, .Where(c => c.TeamId == successor.TeamId && c.ProviderInstanceId == successor.ProviderInstanceId && c.OwnerUserId == ownerId && c.Id != successor.Id) .ToListAsync(cancellationToken).ConfigureAwait(false); - return owned.Where(c => c.Status is CredentialStatus.Revoked or CredentialStatus.Error).Select(c => c.Id).ToList(); + var dead = owned.Where(c => c.Status is CredentialStatus.Revoked or CredentialStatus.Error).ToList(); + + return await HoldAsReadAsync(dead, cancellationToken).ConfigureAwait(false); + } + + /// + /// Locks each dead credential's row until the unit of work ends and keeps only the ones still stored exactly as they were + /// read — same status, same payload. A re-authorization writes the row (status Active, a new payload), so one that landed + /// since the read changed it and the credential is healthy: its bindings stay. One that lands after the lock waits for the + /// commit, because its write needs the row this transaction holds — and then finds the bindings already moved, as it would + /// after the sweep. FOR NO KEY UPDATE excludes that write and nothing that merely refers to the credential; locked in id + /// order so two connects of one owner queue behind each other instead of deadlocking. Compared with the values the entity + /// was loaded with, not the ones it holds, because a token link has already marked the credential it replaces Revoked in memory. + /// + private async Task> HoldAsReadAsync(List dead, CancellationToken cancellationToken) + { + if (dead.Count == 0) return new List(); + + var ids = dead.Select(c => c.Id).ToArray(); + + await HoldRowsAsync(ids, cancellationToken).ConfigureAwait(false); + + var stored = await _db.Credential.AsNoTracking().Where(c => ids.Contains(c.Id)).Select(c => new StoredState(c.Id, c.Status, c.EncryptedPayload)).ToListAsync(cancellationToken).ConfigureAwait(false); + + return dead.Where(credential => IsStoredAsRead(credential, stored)).Select(c => c.Id).ToList(); + } + + /// Locks the rows until the transaction they are taken in ends, in id order so two holders of overlapping sets queue behind each other instead of deadlocking. A row that is not there is not an error: whoever judges it afterwards finds nothing. + private async Task HoldRowsAsync(Guid[] ids, CancellationToken cancellationToken) => + await _db.Database.ExecuteSqlInterpolatedAsync($"SELECT 1 FROM credential WHERE id = ANY({ids}) ORDER BY id FOR NO KEY UPDATE", cancellationToken).ConfigureAwait(false); + + private bool IsStoredAsRead(Credential credential, List stored) + { + var read = _db.Entry(credential).OriginalValues; + + return stored.Any(s => s.Id == credential.Id && s.Status == read.GetValue(nameof(Credential.Status)) && s.EncryptedPayload == read.GetValue(nameof(Credential.EncryptedPayload))); } /// @@ -96,19 +145,23 @@ private async Task> LoadStrandedAsync(Guid? after, int .Take(take) .ToListAsync(cancellationToken).ConfigureAwait(false); - return rows.Select(row => new StrandedCredential(row.DeadId, row.SuccessorId!.Value)).ToList(); + return rows.Select(row => new StrandedCredential(row.DeadId)).ToList(); } /// /// Every candidate after (all of them when null), unbounded: what - /// pages through and counts, so the two cannot disagree about what a candidate is. + /// pages through and counts, so the two cannot disagree about what a candidate is. Narrowed to + /// it is the verdict on that one credential, which asks for again under + /// the credential's lock — the one definition of "dead and stranded", asked twice. /// - private IQueryable StrandedWithSuccessor(Guid? after = null) + private IQueryable StrandedWithSuccessor(Guid? after = null, Guid? only = null) { var deadCredentials = _db.Credential.AsNoTracking().Where(c => (c.Status == CredentialStatus.Revoked || c.Status == CredentialStatus.Error) && c.OwnerUserId != null); if (after is Guid cursor) deadCredentials = deadCredentials.Where(c => c.Id.CompareTo(cursor) > 0); + if (only is Guid single) deadCredentials = deadCredentials.Where(c => c.Id == single); + return deadCredentials .Where(dead => _db.Repository.Any(r => r.CredentialId == dead.Id && r.TeamId == dead.TeamId && r.ProviderInstanceId == dead.ProviderInstanceId && r.DeletedDate == null) || _db.ConnectionWebhook.Any(w => w.CredentialId == dead.Id && w.ProviderInstanceId == dead.ProviderInstanceId && WebhookRegistrationLifecycle.InService.Contains(w.RegistrationStatus))) @@ -124,16 +177,38 @@ private IQueryable StrandedWithSuccessor(Guid? after = null) .Where(candidate => candidate.SuccessorId != null); } - /// Move one dead credential's bindings and save them on their own, then hand its revived hooks to the registrar. + /// + /// Move one dead credential's bindings, under the credential's refresh lock. A credential somebody is writing — a refresh, or + /// a re-authorization that may be reviving it — is skipped, not waited for: nothing is lost, the next pass meets it again. + /// private async Task RepairAsync(StrandedCredential candidate, CancellationToken cancellationToken) + { + await using var held = await CredentialRefreshLock.TryAcquireAsync(_tryLock, candidate.DeadId, cancellationToken).ConfigureAwait(false); + + if (held is null) return SkipBusy(candidate); + + return await RepairHeldAsync(candidate, cancellationToken).ConfigureAwait(false); + } + + private bool SkipBusy(StrandedCredential candidate) + { + _logger.LogInformation("{Sweep}: credential {CredentialId} is being written (a refresh or a re-authorization holds its lock); it stays a candidate for the next pass", SweepName, candidate.DeadId); + + return false; + } + + /// + /// The candidate was read before the lock was taken, so it is judged again now — and the move saved and committed as one + /// transaction (). The revived hooks go to the registrar once it has committed. + /// + private async Task RepairHeldAsync(StrandedCredential candidate, CancellationToken cancellationToken) { try { - var successor = await _db.Credential.SingleAsync(c => c.Id == candidate.SuccessorId, cancellationToken).ConfigureAwait(false); + var adopted = await MoveWhileHoldingRowAsync(candidate, cancellationToken).ConfigureAwait(false); - var adopted = await AdoptBindingsAsync(successor, new List { candidate.DeadId }, cancellationToken).ConfigureAwait(false); + if (adopted is null) return false; - await _db.SaveChangesAsync(cancellationToken).ConfigureAwait(false); await DispatchAfterCommitAsync(adopted.RevivedHookIds, cancellationToken).ConfigureAwait(false); return true; @@ -144,6 +219,34 @@ private async Task RepairAsync(StrandedCredential candidate, CancellationT } } + /// + /// The one credential's own transaction — not the batch's, which a sweep must not have (): the row is + /// locked FOR NO KEY UPDATE first and judged dead and stranded second, so what is moved is judged on the row as it is for as + /// long as the move takes. The refresh lock excludes the writers of the token — a refresh, a re-authorization — and this excludes + /// every other writer of the row, which the hourly health probe is: it writes a refused credential back to Active when its + /// provider accepts it again, without that lock. A verdict recorded since the page was read is seen by the judgement; one that + /// arrives after the row is held waits for the commit, and then finds the bindings already moved. The successor is the one this + /// verdict names, not the one the page did. Null when the credential is no longer a candidate: nothing moves. + /// + private async Task MoveWhileHoldingRowAsync(StrandedCredential candidate, CancellationToken cancellationToken) + { + await using var transaction = await _db.Database.BeginTransactionAsync(cancellationToken).ConfigureAwait(false); + + await HoldRowsAsync(new[] { candidate.DeadId }, cancellationToken).ConfigureAwait(false); + + var verdict = await StrandedWithSuccessor(only: candidate.DeadId).SingleOrDefaultAsync(cancellationToken).ConfigureAwait(false); + + if (verdict is null) return null; + + var successor = await _db.Credential.SingleAsync(c => c.Id == verdict.SuccessorId!.Value, cancellationToken).ConfigureAwait(false); + var adopted = await AdoptBindingsAsync(successor, new List { candidate.DeadId }, cancellationToken).ConfigureAwait(false); + + await _db.SaveChangesAsync(cancellationToken).ConfigureAwait(false); + await transaction.CommitAsync(cancellationToken).ConfigureAwait(false); + + return adopted; + } + /// Re-point every live binding of at as tracked changes. Returns what moved, and the group hooks now waiting for a registrar. private async Task AdoptBindingsAsync(Credential successor, List predecessorIds, CancellationToken cancellationToken) { @@ -216,11 +319,14 @@ private async Task DispatchAfterCommitAsync(List hookIds, CancellationToke await _postCommit.RunAfterCommitAsync(ct => _hookDispatcher.DispatchAsync(hookId, ct), cancellationToken).ConfigureAwait(false); } - private sealed record StrandedCredential(Guid DeadId, Guid SuccessorId) : ISweepCandidate + private sealed record StrandedCredential(Guid DeadId) : ISweepCandidate { public Guid Id => DeadId; } + /// What a credential's row holds that a re-authorization changes: compared with what the entity was read with. + private sealed record StoredState(Guid Id, CredentialStatus Status, string EncryptedPayload); + private sealed record Adopted(int RepositoryCount, int ConnectionWebhookCount, List RevivedHookIds); /// The query's row: a member-initialised class because an anonymous type cannot be returned from . diff --git a/backend/src/CodeSpace.Core/Services/Credentials/ICredentialRevocationService.cs b/backend/src/CodeSpace.Core/Services/Credentials/ICredentialRevocationService.cs index ad183820a..1dd20e537 100644 --- a/backend/src/CodeSpace.Core/Services/Credentials/ICredentialRevocationService.cs +++ b/backend/src/CodeSpace.Core/Services/Credentials/ICredentialRevocationService.cs @@ -38,7 +38,9 @@ public interface ICredentialRevocationService /// tokens, by an older pod, or by an instance Admin losing the role; each exactly as a departure retires it and saved /// on its own. A token that cannot be attributed to an account is never one. Returns how many were retired. A token that /// fails is skipped and logged and the pass goes on past it — however many fail, they cannot crowd a healthy one out — and - /// one that has failed several passes in a row is set aside for a while (). + /// one that has failed several passes in a row is set aside for a while (). Each token is + /// retired under its refresh lock, judged departed again once it is held: a token a refresh or a re-authorization is + /// writing is skipped and met on the next pass, so a retirement never blanks what such a write lands. /// Task RetireDepartedAsync(int batchSize, CancellationToken cancellationToken); } diff --git a/backend/src/CodeSpace.Core/Services/Credentials/ICredentialSuccessionService.cs b/backend/src/CodeSpace.Core/Services/Credentials/ICredentialSuccessionService.cs index c206337d8..ece26c5b6 100644 --- a/backend/src/CodeSpace.Core/Services/Credentials/ICredentialSuccessionService.cs +++ b/backend/src/CodeSpace.Core/Services/Credentials/ICredentialSuccessionService.cs @@ -17,7 +17,9 @@ public interface ICredentialSuccessionService /// , lifting the "needs a new credential" error a disconnect set and reviving the hooks its /// failures buried. A credential with no /// owner (team-service) succeeds nobody. Tracked changes only — the caller's unit of work saves them with the - /// successor, and the revived hooks are dispatched after it commits. + /// successor, and the revived hooks are dispatched after it commits. A dead credential that a re-authorization has + /// revived since it was read keeps its bindings, and one revived after that waits for the unit of work to commit: the + /// credentials are held until it ends. /// Task CarryForwardAsync(Credential successor, CancellationToken cancellationToken); @@ -28,7 +30,10 @@ public interface ICredentialSuccessionService /// credential that fails is skipped and logged, and the pass goes on to the ones after it — however many fail, they /// cannot crowd a repairable credential out of the pass — and one that has failed several passes in a row is set aside /// for a while rather than attempted, and warned about, every tick. A pass is bounded in attempts as well as in - /// repairs (). + /// repairs (). Each credential is repaired under its refresh lock and judged dead + /// again once it is held: one a re-authorization has revived since the page was read is no longer a candidate and keeps + /// its bindings, and one a refresh or a re-authorization is writing is skipped, not waited for, and met on the next pass. The move is made + /// holding the credential's row, so a verdict the health probe records meanwhile waits for it. /// Task RepairStrandedAsync(int batchSize, CancellationToken cancellationToken); diff --git a/backend/src/CodeSpace.Core/Services/OAuth/CredentialRefreshLock.cs b/backend/src/CodeSpace.Core/Services/OAuth/CredentialRefreshLock.cs index e5c8aae6a..57ac9f83d 100644 --- a/backend/src/CodeSpace.Core/Services/OAuth/CredentialRefreshLock.cs +++ b/backend/src/CodeSpace.Core/Services/OAuth/CredentialRefreshLock.cs @@ -5,7 +5,11 @@ namespace CodeSpace.Core.Services.OAuth; /// /// The lock everything that rewrites one credential's token material takes, so two of them never write at once: a refresh /// () and a re-authorization (ICredentialReauthorizer) are the same kind of act — -/// each replaces the stored pair, and each is only sound against the pair it read. Two tiers, always taken in this order: +/// each replaces the stored pair, and each is only sound against the pair it read. The recurring sweeps that decide a +/// credential is dead and then move what is bound to it or retire it take it too (), because a +/// re-authorization that lands between their verdict and their act would otherwise be acted against. It excludes the writers of the +/// token and no others: a repair's move excludes a write of the credential's row that does not take it — the health probe's — +/// by holding the row instead (CredentialSuccessionService). Two tiers, always taken in this order: /// 1. a per-credential in-process — cheap, and blocks every sibling in this process; /// 2. the (a Postgres advisory lock) — blocks every other pod. /// One definition, because the semaphores are process-wide state: a second dictionary beside this one would be a second lock @@ -40,6 +44,35 @@ internal static async Task AcquireAsync(ICrossProcessLock cros } } + /// + /// Both tiers if neither is held — null at once, without waiting, when a write of this credential's token is under way in + /// this process or on another pod. For a sweep, which skips a busy credential and meets it again on its next pass rather + /// than queue behind a provider call; the same semaphore and the same advisory key as , so a sweep + /// that holds a credential excludes a refresh or a re-authorization exactly as they exclude each other. + /// + internal static async Task TryAcquireAsync(ICrossProcessTryLock crossProcess, Guid credentialId, CancellationToken cancellationToken) + { + var semaphore = InProcess.GetOrAdd(credentialId, _ => new SemaphoreSlim(1, 1)); + + if (!await semaphore.WaitAsync(0, cancellationToken).ConfigureAwait(false)) return null; + + try + { + var advisory = await crossProcess.TryAcquireAsync(KeyOf(credentialId), cancellationToken).ConfigureAwait(false); + + if (advisory is not null) return new Held(semaphore, advisory); + } + catch + { + semaphore.Release(); + throw; + } + + semaphore.Release(); + + return null; + } + private sealed class Held : IAsyncDisposable { private readonly SemaphoreSlim _semaphore; diff --git a/backend/src/CodeSpace.Core/Services/OAuth/ICrossProcessTryLock.cs b/backend/src/CodeSpace.Core/Services/OAuth/ICrossProcessTryLock.cs index 5a28ca92b..b72ed334d 100644 --- a/backend/src/CodeSpace.Core/Services/OAuth/ICrossProcessTryLock.cs +++ b/backend/src/CodeSpace.Core/Services/OAuth/ICrossProcessTryLock.cs @@ -2,8 +2,8 @@ namespace CodeSpace.Core.Services.OAuth; /// /// A sibling of for work that must not run twice at once and should simply be skipped, not -/// queued, when something else is already doing it: a recurring sweep whose previous pass is still going. Same key space as -/// . +/// queued, when something else is already doing it: a recurring sweep whose previous pass is still going, or a credential a +/// write of its token holds (CredentialRefreshLock.TryAcquireAsync). Same key space as . /// public interface ICrossProcessTryLock { diff --git a/backend/src/CodeSpace.Messages/Commands/Credentials/RepairDisconnectedCredentialBindingsCommand.cs b/backend/src/CodeSpace.Messages/Commands/Credentials/RepairDisconnectedCredentialBindingsCommand.cs index cd6351e43..7874a4942 100644 --- a/backend/src/CodeSpace.Messages/Commands/Credentials/RepairDisconnectedCredentialBindingsCommand.cs +++ b/backend/src/CodeSpace.Messages/Commands/Credentials/RepairDisconnectedCredentialBindingsCommand.cs @@ -7,7 +7,8 @@ namespace CodeSpace.Messages.Commands.Credentials; /// own owner connected afterwards on the same provider instance and team. The connect paths carry these bindings /// forward themselves; this repairs the ones a connect path left behind anyway — rows stranded before the carry-forward /// existed, a connect served by an older pod during a rolling deploy, and any later path that strands one. Idempotent -/// and bounded: a moved binding no longer points at a revoked credential, so the backlog only shrinks. NOT +/// and bounded: a moved binding no longer points at a revoked credential, so the backlog only shrinks. A credential a refresh +/// or an in-place re-authorization is writing is skipped for this pass, and one revived since it was read keeps its bindings. NOT /// tenant-scoped: a system-wide repair that runs without an actor context (mirrors BackfillPackCloneUrlsCommand). /// /// NOT transactional (): each credential's bindings are saved on their own, diff --git a/backend/src/CodeSpace.Messages/Commands/Credentials/RetireDepartedMembersCredentialsCommand.cs b/backend/src/CodeSpace.Messages/Commands/Credentials/RetireDepartedMembersCredentialsCommand.cs index d2cf49695..2f670c2e6 100644 --- a/backend/src/CodeSpace.Messages/Commands/Credentials/RetireDepartedMembersCredentialsCommand.cs +++ b/backend/src/CodeSpace.Messages/Commands/Credentials/RetireDepartedMembersCredentialsCommand.cs @@ -11,7 +11,8 @@ namespace CodeSpace.Messages.Commands.Credentials; /// token an instance Admin lent a team they are not in, once they lose the role — the app never takes the role away, so /// no departure retires those. Each token is retired exactly as a departure retires it — blanked, Revoked, /// soft-deleted, its repositories marked "needs a new credential", its act-as identities soft-deleted — and the provider -/// is not asked. Idempotent and bounded: a retired token is no longer a candidate, so the backlog only shrinks. NOT +/// is not asked. Idempotent and bounded: a retired token is no longer a candidate, so the backlog only shrinks. A token a +/// refresh or an in-place re-authorization is writing is skipped for this pass and retired by a later one. NOT /// tenant-scoped: a system-wide repair that runs without an actor context (mirrors RepairDisconnectedCredentialBindingsCommand). /// /// NOT transactional (): each token is retired and saved on its own, and a diff --git a/backend/tests/CodeSpace.E2ETests/Credentials/CredentialSweepLockE2ETests.cs b/backend/tests/CodeSpace.E2ETests/Credentials/CredentialSweepLockE2ETests.cs new file mode 100644 index 000000000..9b554e23a --- /dev/null +++ b/backend/tests/CodeSpace.E2ETests/Credentials/CredentialSweepLockE2ETests.cs @@ -0,0 +1,272 @@ +using System.Net; +using System.Text.Json; +using CodeSpace.Core.Persistence.Db; +using CodeSpace.Core.Persistence.Entities; +using CodeSpace.Core.Services.Credentials; +using CodeSpace.Core.Services.OAuth; +using CodeSpace.E2ETests.Infrastructure; +using CodeSpace.IntegrationTests.Infrastructure; +using CodeSpace.Messages.Commands.Credentials; +using CodeSpace.Messages.Constants; +using CodeSpace.Messages.Credentials; +using CodeSpace.Messages.Enums; +using CodeSpace.Messages.Mediation; +using MediatR; +using Microsoft.EntityFrameworkCore; +using Microsoft.Extensions.DependencyInjection; +using Shouldly; + +namespace CodeSpace.E2ETests.Credentials; + +/// +/// 🟢 High fidelity — the two recurring sweeps that act on a credential they have judged dead, run against a credential whose +/// token somebody is writing: the repair of stranded bindings () and +/// the retirement of a departed member's tokens (), each sent through the host's own +/// mediator — the Hangfire tick that would send them is not what is under test. The re-authorization is the real HTTP one (the +/// GitLab callback, through the real ASP.NET pipeline and the real GitLab OAuth client), against real Postgres and a +/// on a loopback port. The one stand-in is the writer's lock: another pod's hold on the +/// credential's advisory lock is taken by the test on a connection of its own, as a second host would take it. +/// +/// What this pins end to end: a person's reconnect that is queued behind a write is not undone by a repair pass that +/// runs meanwhile — the pass skips the credential, the reconnect then lands, and the next pass finds a healthy credential and +/// leaves its repository on it; and a departed member's token that is being written is kept by the pass that finds it busy and +/// retired by the one after, the skipped passes counting as nothing against it. The interleavings inside one request (a +/// verdict recorded between a sweep's read and its move) are pinned by the integration tier, where the gap can be made. +/// +[Trait("Category", "E2E")] +[Trait("Category", "ProviderE2E")] +[Trait("Surface", "Http")] +public sealed class CredentialSweepLockE2ETests : IClassFixture +{ + private const string BindingCookie = "codespace_oauth_nonce"; + private const string CallbackPath = "/api/credentials/oauth/callback"; + + private static readonly TimeSpan Patience = TimeSpan.FromSeconds(30); + + private readonly TaskLaunchApiFactory _factory; + private readonly TenantHarness _tenants; + + public CredentialSweepLockE2ETests(TaskLaunchApiFactory factory) + { + _factory = factory; + _tenants = new TenantHarness(factory); + } + + [Fact] + public async Task A_repair_pass_that_runs_while_a_reconnect_is_waiting_to_write_moves_nothing_and_the_next_one_leaves_the_reconnected_credential_its_repository() + { + using var gitlab = new GitLabDoorkeeperStub(); + var world = await SeedStrandedAsync(gitlab).ConfigureAwait(false); + var init = await InitAsync(world).ConfigureAwait(false); + var state = await StateOfAsync(init).ConfigureAwait(false); + Task reconnect; + + await using (await HoldLockLikeAnotherPodAsync(world.DeadId).ConfigureAwait(false)) + { + reconnect = Task.Run(() => CallbackAsync(state, "authz:nacho", CookieValue(BindingCookieOf(init)!))); + await WaitForAsync(AnAdvisoryLockIsWaitedForAsync, $"the reconnect, having exchanged its code, queued for credential {world.DeadId}'s write lock — `SELECT * FROM pg_locks WHERE locktype = 'advisory' AND NOT granted` shows whether it did").ConfigureAwait(false); + reconnect.IsCompleted.ShouldBeFalse("fixture check: the reconnect is waiting for the write lock another pod holds"); + + await RepairWithinPatienceAsync("the repair pass finished without waiting for the lock").ConfigureAwait(false); + + (await ReadRepositoryAsync(world.RepositoryId).ConfigureAwait(false)).CredentialId.ShouldBe(world.DeadId, "the credential is being written: the pass cannot know the reconnect is about to revive it, so it moved nothing"); + (await ReadCredentialAsync(world.DeadId).ConfigureAwait(false)).Status.ShouldBe(CredentialStatus.Error, "fixture check: the reconnect has not landed yet"); + } + + var callback = await reconnect.WaitAsync(Patience).ConfigureAwait(false); + callback.StatusCode.ShouldBe(HttpStatusCode.Redirect, await TenantHarness.DescribeAsync(callback).ConfigureAwait(false)); + (await ReadCredentialAsync(world.DeadId).ConfigureAwait(false)).Status.ShouldBe(CredentialStatus.Active, "the reconnect landed once the lock was free"); + + await RepairWithinPatienceAsync("the repair pass after the reconnect finished").ConfigureAwait(false); + + (await ReadRepositoryAsync(world.RepositoryId).ConfigureAwait(false)).ShouldSatisfyAllConditions(r => r.CredentialId.ShouldBe(world.DeadId, "healthy again, so no longer stranded: its repository stays on it"), r => r.Status.ShouldBe(RepositoryStatus.Active)); + (await _tenants.DbAsync(db => db.Repository.AsNoTracking().CountAsync(r => r.CredentialId == world.SuccessorId)).ConfigureAwait(false)).ShouldBe(0, "nothing was carried onto the owner's other credential"); + await AssertLockFreeAsync(world.DeadId).ConfigureAwait(false); + } + + [Fact] + public async Task A_repair_pass_that_finds_nothing_held_moves_a_stranded_credentials_repository_to_the_owners_live_one_and_gives_the_lock_back() + { + using var gitlab = new GitLabDoorkeeperStub(); + var world = await SeedStrandedAsync(gitlab).ConfigureAwait(false); + + await RepairWithinPatienceAsync("the repair pass finished").ConfigureAwait(false); + + (await ReadRepositoryAsync(world.RepositoryId).ConfigureAwait(false)).CredentialId.ShouldBe(world.SuccessorId, "the normal case, as it always was: a credential nobody is writing and whose owner connected another is repaired"); + await AssertLockFreeAsync(world.DeadId).ConfigureAwait(false); + } + + [Fact] + public async Task A_departed_members_token_that_is_being_written_is_kept_by_every_pass_that_finds_it_busy_and_retired_by_the_first_that_does_not() + { + var world = await SeedDepartedMembersTokenAsync().ConfigureAwait(false); + + await using (await HoldLockLikeAnotherPodAsync(world.CredentialId).ConfigureAwait(false)) + { + for (var pass = 0; pass <= SweepFailureBackoff.FailuresBeforeBackoff; pass++) + await RetireWithinPatienceAsync("the retirement pass finished without waiting for the lock").ConfigureAwait(false); + + (await ReadCredentialAsync(world.CredentialId).ConfigureAwait(false)).ShouldSatisfyAllConditions(c => c.EncryptedPayload.ShouldNotBeEmpty("somebody is writing the token: a retirement under that write would blank what it lands"), c => c.DeletedDate.ShouldBeNull()); + (await ReadRepositoryAsync(world.RepositoryId).ConfigureAwait(false)).Status.ShouldBe(RepositoryStatus.Active); + } + + await RetireWithinPatienceAsync("the retirement pass after the write finished").ConfigureAwait(false); + + (await ReadCredentialAsync(world.CredentialId).ConfigureAwait(false)).ShouldSatisfyAllConditions(c => c.EncryptedPayload.ShouldBeEmpty("a busy token is not a failing one: the passes that skipped it did not set it aside"), c => c.DeletedDate.ShouldNotBeNull(), c => c.Status.ShouldBe(CredentialStatus.Revoked)); + (await ReadRepositoryAsync(world.RepositoryId).ConfigureAwait(false)).ShouldSatisfyAllConditions(r => r.Status.ShouldBe(RepositoryStatus.Error), r => r.LastError.ShouldBe(CredentialRevocationService.DisconnectedRepositoryError)); + await AssertLockFreeAsync(world.CredentialId).ConfigureAwait(false); + } + + // ─── Plumbing ────────────────────────────────────────────────────────────── + + private sealed record StrandedWorld(Guid OwnerId, Guid TeamId, Guid InstanceId, Guid DeadId, Guid SuccessorId, Guid RepositoryId); + + private sealed record DepartedWorld(Guid CredentialId, Guid RepositoryId); + + /// + /// What the repair is for: nacho's credential refused by GitLab, with a repository bound to it, and a live credential of his + /// connected after it to follow. The dead one has an id that sorts ahead of any random one, so a pass reads it in its first page + /// whatever else the shared database holds. + /// + private async Task SeedStrandedAsync(GitLabDoorkeeperStub gitlab) + { + var owner = await _tenants.AccountAsync().ConfigureAwait(false); + var team = await _tenants.TeamAsync((owner, TeamRole.Admin)).ConfigureAwait(false); + + using var scope = _factory.Services.CreateScope(); + var db = scope.ServiceProvider.GetRequiredService(); + var encryptor = scope.ServiceProvider.GetRequiredService(); + var serializer = scope.ServiceProvider.GetRequiredService(); + + var instance = new ProviderInstance { Id = Guid.NewGuid(), TeamId = team, Provider = ProviderKind.GitLab, DisplayName = $"gitlab-{Guid.NewGuid():N}"[..14], BaseUrl = gitlab.BaseUrl, ApiUrl = gitlab.BaseUrl, OauthClientId = GitLabDoorkeeperStub.DefaultClientId, OauthClientSecretEnc = encryptor.Encrypt(GitLabDoorkeeperStub.DefaultClientSecret), CreatedBy = SystemUsers.SeederId, LastModifiedBy = SystemUsers.SeederId }; + var dead = NachosCredential(team, instance.Id, owner, encryptor, serializer, "at-0", "rt-0", FrontOfTheQueue(), CredentialStatus.Error, "401 Unauthorized"); + var repository = new Repository { Id = Guid.NewGuid(), TeamId = team, ProviderInstanceId = instance.Id, CredentialId = dead.Id, ExternalId = Random.Shared.Next(100_000, 999_999).ToString(), NamespacePath = "acme", Name = "api", FullPath = "acme/api", DefaultBranch = "main", Visibility = RepositoryVisibility.Private, WebUrl = $"{gitlab.BaseUrl}/acme/api", CloneUrlHttps = $"{gitlab.BaseUrl}/acme/api.git", Status = RepositoryStatus.Active }; + var identity = new UserProviderIdentity { Id = Guid.NewGuid(), UserId = owner, ProviderInstanceId = instance.Id, CredentialId = dead.Id, ProviderUserId = gitlab.UserIdOf("nacho").ToString(), ProviderUsername = "nacho" }; + + db.ProviderInstance.Add(instance); + db.Credential.Add(dead); + db.Repository.Add(repository); + db.UserProviderIdentity.Add(identity); + await db.SaveChangesAsync().ConfigureAwait(false); + + var successor = NachosCredential(team, instance.Id, owner, encryptor, serializer, "at-1", "rt-1", Guid.NewGuid(), CredentialStatus.Active, lastError: null); + + db.Credential.Add(successor); + await db.SaveChangesAsync().ConfigureAwait(false); + + return new StrandedWorld(owner, team, instance.Id, dead.Id, successor.Id, repository.Id); + } + + private static Credential NachosCredential(Guid team, Guid instanceId, Guid owner, IPayloadEncryptor encryptor, ICredentialPayloadSerializer serializer, string accessToken, string refreshToken, Guid id, CredentialStatus status, string? lastError) + { + var payload = new OAuthPayload { AccessToken = accessToken, RefreshToken = refreshToken, ExpiresAt = DateTimeOffset.UtcNow.AddMinutes(status == CredentialStatus.Error ? -1 : 120) }; + + return new Credential { Id = id, TeamId = team, ProviderInstanceId = instanceId, OwnerUserId = owner, Ownership = CredentialOwnership.Personal, AuthType = payload.Type, DisplayName = "GitLab · nacho", EncryptedPayload = encryptor.Encrypt(serializer.Serialize(payload)), ExpiresDate = payload.ExpiresAt, Status = status, LastError = lastError, CreatedBy = owner, LastModifiedBy = owner }; + } + + /// What the retirement is for: a live personal token of someone who never joined the team, with a repository bound through it. + private async Task SeedDepartedMembersTokenAsync() + { + var admin = await _tenants.AccountAsync().ConfigureAwait(false); + var departed = await _tenants.AccountAsync().ConfigureAwait(false); + var team = await _tenants.TeamAsync((admin, TeamRole.Admin)).ConfigureAwait(false); + + using var scope = _factory.Services.CreateScope(); + var db = scope.ServiceProvider.GetRequiredService(); + var encryptor = scope.ServiceProvider.GetRequiredService(); + var serializer = scope.ServiceProvider.GetRequiredService(); + + var instance = new ProviderInstance { Id = Guid.NewGuid(), TeamId = team, Provider = ProviderKind.GitLab, DisplayName = $"gitlab-{Guid.NewGuid():N}"[..14], BaseUrl = "http://127.0.0.1:1", CreatedBy = SystemUsers.SeederId, LastModifiedBy = SystemUsers.SeederId }; + var payload = new PatPayload { Token = $"glpat-departed-{Guid.NewGuid():N}" }; + var credential = new Credential { Id = FrontOfTheQueue(), TeamId = team, ProviderInstanceId = instance.Id, OwnerUserId = departed, Ownership = CredentialOwnership.Personal, AuthType = payload.Type, DisplayName = "GitLab · departed", EncryptedPayload = encryptor.Encrypt(serializer.Serialize(payload)), Status = CredentialStatus.Active, CreatedBy = departed, LastModifiedBy = departed }; + var repository = new Repository { Id = Guid.NewGuid(), TeamId = team, ProviderInstanceId = instance.Id, CredentialId = credential.Id, ExternalId = Random.Shared.Next(100_000, 999_999).ToString(), NamespacePath = "acme", Name = "api", FullPath = "acme/api", DefaultBranch = "main", Visibility = RepositoryVisibility.Private, WebUrl = "http://127.0.0.1:1/acme/api", CloneUrlHttps = "http://127.0.0.1:1/acme/api.git", Status = RepositoryStatus.Active }; + + db.ProviderInstance.Add(instance); + db.Credential.Add(credential); + db.Repository.Add(repository); + await db.SaveChangesAsync().ConfigureAwait(false); + + return new DepartedWorld(credential.Id, repository.Id); + } + + /// An id that sorts ahead of any random one: a sweep reads its candidates in id order, so a candidate here is in the first page whatever else the shared database holds. + private static Guid FrontOfTheQueue() => Guid.Parse($"00000000-0000-0000-0000-{Guid.NewGuid().ToString("N")[..12]}"); + + private async Task InitAsync(StrandedWorld world) => + await _tenants.Client().SendAsync(_tenants.Request(HttpMethod.Post, "/api/credentials/oauth/init", world.OwnerId, world.TeamId, new { providerInstanceId = world.InstanceId, displayName = "GitLab · nacho", returnUrl = "/settings/credentials", replaceCredentialId = world.DeadId })).ConfigureAwait(false); + + private async Task CallbackAsync(string state, string code, string bindingCookie) + { + var request = _tenants.Request(HttpMethod.Get, $"{CallbackPath}?code={Uri.EscapeDataString(code)}&state={Uri.EscapeDataString(state)}", userId: null, teamId: null); + + request.Headers.Add("Cookie", $"{BindingCookie}={bindingCookie}"); + + return await _tenants.Client().SendAsync(request).ConfigureAwait(false); + } + + /// What another pod holds while it writes the credential's token: the advisory lock the write takes, on a connection that is not this host's. + private async Task HoldLockLikeAnotherPodAsync(Guid credentialId) + { + using var scope = _factory.Services.CreateScope(); + + return await scope.ServiceProvider.GetRequiredService().AcquireAsync(CredentialRefreshLock.KeyOf(credentialId), CancellationToken.None).ConfigureAwait(false); + } + + /// The lock was given back: both tiers are taken at once. A lock a sweep leaked would hang every later write of the credential, so this waits only as long as a free lock takes. + private async Task AssertLockFreeAsync(Guid credentialId) + { + using var scope = _factory.Services.CreateScope(); + + await using var held = await CredentialRefreshLock.AcquireAsync(scope.ServiceProvider.GetRequiredService(), credentialId, CancellationToken.None).WaitAsync(TimeSpan.FromSeconds(5)).ConfigureAwait(false); + } + + private async Task RepairWithinPatienceAsync(string signal) => + await SweepWithinPatienceAsync(new RepairDisconnectedCredentialBindingsCommand { BatchSize = 500 }, signal).ConfigureAwait(false); + + private async Task RetireWithinPatienceAsync(string signal) => + await SweepWithinPatienceAsync(new RetireDepartedMembersCredentialsCommand { BatchSize = 500 }, signal).ConfigureAwait(false); + + private async Task SweepWithinPatienceAsync(ICommand sweep, string signal) + { + using var scope = _factory.Services.CreateScope(); + + try + { + await scope.ServiceProvider.GetRequiredService().Send(sweep).WaitAsync(Patience).ConfigureAwait(false); + } + catch (TimeoutException timeout) + { + throw new TimeoutException($"{signal} within {Patience.TotalSeconds}s — a sweep waits for no lock, so look for a transaction holding a row it needs: `SELECT pid, pg_blocking_pids(pid), state, query FROM pg_stat_activity WHERE datname = current_database()`.", timeout); + } + } + + /// True while some session in this database is queued for an advisory lock it has not been granted: the reconnect waiting for the write lock another pod holds. + private async Task AnAdvisoryLockIsWaitedForAsync() => + await _tenants.DbAsync(async db => await db.Database.SqlQuery($"SELECT count(*)::int AS \"Value\" FROM pg_locks WHERE locktype = 'advisory' AND NOT granted AND database = (SELECT oid FROM pg_database WHERE datname = current_database())").SingleAsync().ConfigureAwait(false) > 0).ConfigureAwait(false); + + private static async Task WaitForAsync(Func> condition, string signal) + { + var deadline = DateTimeOffset.UtcNow + Patience; + + while (!await condition().ConfigureAwait(false)) + { + if (DateTimeOffset.UtcNow > deadline) throw new TimeoutException($"{signal} — not seen within {Patience.TotalSeconds}s"); + + await Task.Delay(50).ConfigureAwait(false); + } + } + + private async Task ReadCredentialAsync(Guid credentialId) => + await _tenants.DbAsync(db => db.Credential.AsNoTracking().SingleAsync(c => c.Id == credentialId)).ConfigureAwait(false); + + private async Task ReadRepositoryAsync(Guid repositoryId) => + await _tenants.DbAsync(db => db.Repository.AsNoTracking().SingleAsync(r => r.Id == repositoryId)).ConfigureAwait(false); + + private static string? BindingCookieOf(HttpResponseMessage response) => + response.Headers.TryGetValues("Set-Cookie", out var values) ? values.FirstOrDefault(c => c.StartsWith($"{BindingCookie}=", StringComparison.Ordinal)) : null; + + private static string CookieValue(string setCookie) => setCookie.Split(';')[0][(BindingCookie.Length + 1)..]; + + private static async Task StateOfAsync(HttpResponseMessage init) => + JsonDocument.Parse(await init.Content.ReadAsStringAsync().ConfigureAwait(false)).RootElement.GetProperty("state").GetString()!; +} diff --git a/backend/tests/CodeSpace.IntegrationTests/Credentials/CredentialReauthorizationFlowTests.Sweeps.cs b/backend/tests/CodeSpace.IntegrationTests/Credentials/CredentialReauthorizationFlowTests.Sweeps.cs new file mode 100644 index 000000000..62bf7bdda --- /dev/null +++ b/backend/tests/CodeSpace.IntegrationTests/Credentials/CredentialReauthorizationFlowTests.Sweeps.cs @@ -0,0 +1,488 @@ +using System.Data.Common; +using Autofac; +using CodeSpace.Core.Persistence.Db; +using CodeSpace.Core.Persistence.Entities; +using CodeSpace.Core.Services.Credentials; +using CodeSpace.Core.Services.OAuth; +using CodeSpace.IntegrationTests.Infrastructure; +using CodeSpace.Messages.Commands.Credentials; +using CodeSpace.Messages.Enums; +using CodeSpace.Messages.Exceptions; +using MediatR; +using Microsoft.EntityFrameworkCore; +using Microsoft.EntityFrameworkCore.Diagnostics; +using Shouldly; + +namespace CodeSpace.IntegrationTests.Credentials; + +/// +/// Where an in-place re-authorization meets the three things that move or end what hangs on a dead credential: the recurring +/// repair of stranded bindings, the connect-time carry-forward, and the recurring retirement of a departed member's tokens. +/// Each decides a credential is dead, and then acts on that verdict a moment later — so a re-authorization that lands in +/// between used to be acted against: its bindings moved away from a credential that was healthy again, or its new token +/// blanked. The claim is that the verdict is taken again under the lock every rewrite of a credential's token takes, so the +/// two can no longer interleave: whichever comes second finds what the first did. +/// +/// 🟢 High fidelity — the real sweeps through the real mediator pipeline, the real re-authorization (a Reconnect, a pasted +/// token) with its refresh lock, and real Postgres. The gap is made, not hoped for: a command interceptor on the sweep's (or +/// the connect's) own database context runs the person's re-authorization right after the statement that read the candidate, +/// which is the one moment the verdict is stale. +/// +public partial class CredentialReauthorizationFlowTests +{ + public enum DeadAs { Refused, Disconnected } + + public enum HeldBy { ThisProcess, AnotherPod } + + private static readonly TimeSpan LongEnoughToHaveFinished = TimeSpan.FromSeconds(1.5); + + /// Passes that would set a candidate aside had each of them counted as its failure: a credential somebody is writing is skipped, which is not one. + private const int PassesThatWouldSetAFailingCandidateAside = SweepFailureBackoff.FailuresBeforeBackoff + 1; + + // ─── The repair sweep ────────────────────────────────────────────────────────────────────────────────── + + [Theory] + [InlineData(DeadAs.Refused)] + [InlineData(DeadAs.Disconnected)] + public async Task A_re_authorization_that_lands_between_the_sweeps_read_and_its_move_leaves_the_bindings_on_the_credential(DeadAs deadAs) + { + using var gitlab = new GitLabDoorkeeperStub(); + var world = await SeedWorldAsync(gitlab).ConfigureAwait(false); + var stranded = await SeedStrandedAsync(world, deadAs).ConfigureAwait(false); + var landed = new AfterReadOf(ReadsStrandedCandidates, () => ReconnectAsync(world, world.NachoId, stranded.Dead, "authz:nacho")); + + await SweepThroughAsync(landed).ConfigureAwait(false); + + landed.Ran.ShouldBeTrue("fixture check: the re-authorization landed after the sweep had read its candidates"); + (await LoadCredentialAsync(stranded.Dead).ConfigureAwait(false)).Status.ShouldBe(CredentialStatus.Active, "fixture check: the credential is healthy again"); + (await LoadRepositoryAsync(stranded.Repository).ConfigureAwait(false)).ShouldSatisfyAllConditions(r => r.CredentialId.ShouldBe(stranded.Dead, "the sweep judged a credential that was no longer dead, and so moved nothing"), r => r.Status.ShouldBe(RepositoryStatus.Active)); + (await LoadHookAsync(stranded.Hook).ConfigureAwait(false)).CredentialId.ShouldBe(stranded.Dead, "the group hook stays with its credential, revived by the re-authorization"); + await AssertNothingBoundAsync(stranded.Successor).ConfigureAwait(false); + } + + [Fact] + public async Task A_re_authorization_that_arrives_while_the_sweep_is_moving_waits_for_it_and_lands_after_the_move() + { + // The credential was dead when the sweep took its lock and judged it, so the move is right; what the re-authorization must + // not do is write the credential's token while the move is being made and be left looking healthy and unbound by accident. + using var gitlab = new GitLabDoorkeeperStub(); + var world = await SeedWorldAsync(gitlab).ConfigureAwait(false); + var stranded = await SeedStrandedAsync(world, DeadAs.Disconnected).ConfigureAwait(false); + var init = await InitAsync(world, world.NachoId, stranded.Dead).ConfigureAwait(false); + Task? reconnect = null; + var finishedWhileMoving = false; + var arrives = new AfterReadOf(ReadsStrandedCandidates, async () => + { + reconnect = Task.Run(() => CompleteAsync(world, init, "authz:nacho")); + await Task.Delay(LongEnoughToHaveFinished).ConfigureAwait(false); + finishedWhileMoving = reconnect.IsCompleted; + }, nth: 2); + + await SweepThroughAsync(arrives).ConfigureAwait(false); + + arrives.Ran.ShouldBeTrue("fixture check: the sweep judged the credential again under its lock, and the re-authorization arrived right after"); + finishedWhileMoving.ShouldBeFalse("the re-authorization takes the lock the sweep holds while it moves, so it cannot finish inside the move"); + await WithinPatienceAsync(reconnect!, "the re-authorization that waited for the move finished", stranded.Dead).ConfigureAwait(false); + + (await LoadRepositoryAsync(stranded.Repository).ConfigureAwait(false)).ShouldSatisfyAllConditions(r => r.CredentialId.ShouldBe(stranded.Successor, "judged dead under the lock, so it moved"), r => r.Status.ShouldBe(RepositoryStatus.Active)); + (await LoadHookAsync(stranded.Hook).ConfigureAwait(false)).CredentialId.ShouldBe(stranded.Successor); + (await LoadCredentialAsync(stranded.Dead).ConfigureAwait(false)).Status.ShouldBe(CredentialStatus.Active, "the re-authorization then landed on the credential, which has nothing bound to it any more"); + } + + [Fact] + public async Task A_provider_accepting_a_refused_credential_again_before_the_sweeps_move_leaves_the_bindings_on_it() + { + // The health probe rechecks a refused credential every hour, and writes it back to Active without the refresh lock: it is + // not a write of the token, so the lock cannot exclude it, and the verdict taken again once the credential is held has to. + using var gitlab = new GitLabDoorkeeperStub(); + var world = await SeedWorldAsync(gitlab).ConfigureAwait(false); + var stranded = await SeedStrandedAsync(world, DeadAs.Refused).ConfigureAwait(false); + var accepted = false; + var landed = new AfterReadOf(ReadsStrandedCandidates, async () => accepted = await AcceptAgainAsync(stranded.Dead).ConfigureAwait(false)); + + await SweepThroughAsync(landed).ConfigureAwait(false); + + landed.Ran.ShouldBeTrue("fixture check: the provider accepted the token again after the sweep had read its candidates"); + accepted.ShouldBeTrue("fixture check: the verdict was recorded"); + (await LoadCredentialAsync(stranded.Dead).ConfigureAwait(false)).Status.ShouldBe(CredentialStatus.Active, "fixture check: the credential is healthy again"); + (await LoadRepositoryAsync(stranded.Repository).ConfigureAwait(false)).ShouldSatisfyAllConditions(r => r.CredentialId.ShouldBe(stranded.Dead, "the sweep judged a credential its provider had accepted again, and so moved nothing"), r => r.Status.ShouldBe(RepositoryStatus.Active)); + (await LoadHookAsync(stranded.Hook).ConfigureAwait(false)).CredentialId.ShouldBe(stranded.Dead); + await AssertNothingBoundAsync(stranded.Successor).ConfigureAwait(false); + } + + [Fact] + public async Task A_provider_accepting_a_refused_credential_while_the_sweep_is_moving_waits_for_the_move_and_lands_after_it() + { + // The credential was dead when the sweep judged it under its lock, so the move is right. What the acceptance must not do is + // land inside the move, between the verdict and the save, and leave a healthy credential whose bindings have gone. + using var gitlab = new GitLabDoorkeeperStub(); + var world = await SeedWorldAsync(gitlab).ConfigureAwait(false); + var stranded = await SeedStrandedAsync(world, DeadAs.Refused).ConfigureAwait(false); + Task? accepted = null; + var finishedWhileMoving = false; + var arrives = new AfterReadOf(ReadsStrandedCandidates, async () => + { + accepted = Task.Run(() => AcceptAgainAsync(stranded.Dead)); + await Task.Delay(LongEnoughToHaveFinished).ConfigureAwait(false); + finishedWhileMoving = accepted.IsCompleted; + }, nth: 2); + + await SweepThroughAsync(arrives).ConfigureAwait(false); + + arrives.Ran.ShouldBeTrue("fixture check: the sweep judged the credential again under its lock, and the acceptance arrived right after"); + finishedWhileMoving.ShouldBeFalse("the sweep holds the credential's row while it moves, so a write of that row has to wait for the commit"); + (await WithinPatienceAsync(accepted!, "the acceptance that waited for the move finished", stranded.Dead).ConfigureAwait(false)).ShouldBeTrue("it landed once the move had committed"); + + (await LoadRepositoryAsync(stranded.Repository).ConfigureAwait(false)).ShouldSatisfyAllConditions(r => r.CredentialId.ShouldBe(stranded.Successor, "judged dead under the lock, so it moved"), r => r.Status.ShouldBe(RepositoryStatus.Active)); + (await LoadHookAsync(stranded.Hook).ConfigureAwait(false)).CredentialId.ShouldBe(stranded.Successor); + (await LoadCredentialAsync(stranded.Dead).ConfigureAwait(false)).Status.ShouldBe(CredentialStatus.Active, "the acceptance then landed on the credential, which has nothing bound to it any more"); + } + + [Theory] + [InlineData(HeldBy.ThisProcess)] + [InlineData(HeldBy.AnotherPod)] + public async Task A_credential_whose_lock_is_held_is_skipped_by_the_sweep_and_repaired_on_a_later_pass(HeldBy heldBy) + { + using var gitlab = new GitLabDoorkeeperStub(); + var world = await SeedWorldAsync(gitlab).ConfigureAwait(false); + var stranded = await SeedStrandedAsync(world, DeadAs.Disconnected).ConfigureAwait(false); + + var write = await HoldLockOfAsync(stranded.Dead, heldBy).ConfigureAwait(false); + + try + { + for (var pass = 0; pass < PassesThatWouldSetAFailingCandidateAside; pass++) + await WithinPatienceAsync(SweepThroughAsync(interceptor: null), "the sweep finished its pass without waiting for the lock", stranded.Dead).ConfigureAwait(false); + + (await LoadRepositoryAsync(stranded.Repository).ConfigureAwait(false)).CredentialId.ShouldBe(stranded.Dead, "somebody is writing the credential's token right now: its verdict cannot be trusted, so nothing moved"); + (await LoadHookAsync(stranded.Hook).ConfigureAwait(false)).CredentialId.ShouldBe(stranded.Dead); + } + finally + { + await write.DisposeAsync().ConfigureAwait(false); + } + + await SweepThroughAsync(interceptor: null).ConfigureAwait(false); + + (await LoadRepositoryAsync(stranded.Repository).ConfigureAwait(false)).ShouldSatisfyAllConditions(r => r.CredentialId.ShouldBe(stranded.Successor, "a credential that was busy is repaired once nothing holds it: skipping it was not a failure, however many passes it was skipped"), r => r.Status.ShouldBe(RepositoryStatus.Active)); + (await LoadHookAsync(stranded.Hook).ConfigureAwait(false)).CredentialId.ShouldBe(stranded.Successor); + } + + [Fact] + public async Task A_stranded_credential_nobody_is_writing_is_repaired_and_the_lock_is_given_back_for_the_next_re_authorization() + { + using var gitlab = new GitLabDoorkeeperStub(); + var world = await SeedWorldAsync(gitlab).ConfigureAwait(false); + var stranded = await SeedStrandedAsync(world, DeadAs.Refused).ConfigureAwait(false); + + await SweepThroughAsync(interceptor: null).ConfigureAwait(false); + + (await LoadRepositoryAsync(stranded.Repository).ConfigureAwait(false)).CredentialId.ShouldBe(stranded.Successor, "the normal case: nothing holds the credential, so the sweep repairs it as it always did"); + await AssertLockFreeAsync(stranded.Dead).ConfigureAwait(false); + + await WithinPatienceAsync(ReconnectAsync(world, world.NachoId, stranded.Dead, "authz:nacho"), "a re-authorization after the repair took the credential's lock", stranded.Dead).ConfigureAwait(false); + + (await LoadCredentialAsync(stranded.Dead).ConfigureAwait(false)).Status.ShouldBe(CredentialStatus.Active); + (await LoadRepositoryAsync(stranded.Repository).ConfigureAwait(false)).CredentialId.ShouldBe(stranded.Successor, "and what the repair moved stays moved"); + } + + // ─── The connect-time carry-forward ──────────────────────────────────────────────────────────────────── + + [Fact] + public async Task A_re_authorization_that_lands_after_a_connect_read_the_dead_credential_leaves_its_bindings_on_it() + { + using var gitlab = new GitLabDoorkeeperStub(); + var world = await SeedWorldAsync(gitlab).ConfigureAwait(false); + var stranded = await SeedBoundRefusedCredentialAsync(world).ConfigureAwait(false); + var init = await InitAsync(world, world.NachoId, replaceCredentialId: null).ConfigureAwait(false); + var landed = new AfterReadOf(ReadsOwnersCredentials, () => ReconnectAsync(world, world.NachoId, stranded.Dead, "authz:nacho")); + + var connected = await CompleteAsync(world, init, "authz:nacho", Through(landed)).ConfigureAwait(false); + + landed.Ran.ShouldBeTrue("fixture check: the re-authorization landed after the connect read the owner's credentials"); + connected.CredentialId.ShouldNotBe(stranded.Dead); + (await LoadCredentialAsync(stranded.Dead).ConfigureAwait(false)).Status.ShouldBe(CredentialStatus.Active, "fixture check: the credential is healthy again"); + (await LoadRepositoryAsync(stranded.Repository).ConfigureAwait(false)).ShouldSatisfyAllConditions(r => r.CredentialId.ShouldBe(stranded.Dead, "the connect carried forward a credential that was no longer dead — it must not"), r => r.Status.ShouldBe(RepositoryStatus.Active)); + (await LoadHookAsync(stranded.Hook).ConfigureAwait(false)).CredentialId.ShouldBe(stranded.Dead); + await AssertNothingBoundAsync(connected.CredentialId!.Value).ConfigureAwait(false); + } + + [Fact] + public async Task A_re_authorization_that_arrives_while_a_connect_is_carrying_bindings_forward_waits_for_it_to_commit() + { + // A lock held to the end of the request would outlive a rollback and strand every later write of the credential; the row + // the connect read is held by the transaction instead, which ends with it whichever way it ends. + using var gitlab = new GitLabDoorkeeperStub(); + var world = await SeedWorldAsync(gitlab).ConfigureAwait(false); + var stranded = await SeedBoundRefusedCredentialAsync(world).ConfigureAwait(false); + var connect = await InitAsync(world, world.NachoId, replaceCredentialId: null).ConfigureAwait(false); + var reauthorize = await InitAsync(world, world.NachoId, stranded.Dead).ConfigureAwait(false); + Task? reconnect = null; + var finishedBeforeTheCommit = false; + var arrives = new AfterReadOf(ReadsRepositories, async () => + { + reconnect = Task.Run(() => CompleteAsync(world, reauthorize, "authz:nacho")); + await Task.Delay(LongEnoughToHaveFinished).ConfigureAwait(false); + finishedBeforeTheCommit = reconnect.IsCompleted; + }); + + var connected = await CompleteAsync(world, connect, "authz:nacho", Through(arrives)).ConfigureAwait(false); + + arrives.Ran.ShouldBeTrue("fixture check: the re-authorization arrived while the connect was moving the bindings"); + finishedBeforeTheCommit.ShouldBeFalse("the connect holds the credential's row until it commits, so the re-authorization's write has to wait for it"); + await WithinPatienceAsync(reconnect!, "the re-authorization that waited for the commit finished", stranded.Dead).ConfigureAwait(false); + + (await LoadRepositoryAsync(stranded.Repository).ConfigureAwait(false)).ShouldSatisfyAllConditions(r => r.CredentialId.ShouldBe(connected.CredentialId, "the credential was dead when the connect committed, so the move stands"), r => r.Status.ShouldBe(RepositoryStatus.Active)); + (await LoadHookAsync(stranded.Hook).ConfigureAwait(false)).CredentialId.ShouldBe(connected.CredentialId!.Value); + (await LoadCredentialAsync(stranded.Dead).ConfigureAwait(false)).Status.ShouldBe(CredentialStatus.Active, "the re-authorization then landed on the credential"); + } + + // ─── The departed-member retirement ──────────────────────────────────────────────────────────────────── + + [Theory] + [InlineData(HeldBy.ThisProcess)] + [InlineData(HeldBy.AnotherPod)] + public async Task A_departed_members_token_whose_lock_is_held_is_not_retired_until_the_write_is_done(HeldBy heldBy) + { + using var gitlab = new GitLabDoorkeeperStub(); + var world = await SeedWorldAsync(gitlab).ConfigureAwait(false); + var (credentialId, repositoryId) = await SeedDepartedMembersTokenAsync(world).ConfigureAwait(false); + + var write = await HoldLockOfAsync(credentialId, heldBy).ConfigureAwait(false); + + try + { + for (var pass = 0; pass < PassesThatWouldSetAFailingCandidateAside; pass++) + await WithinPatienceAsync(RetireThroughAsync(interceptor: null), "the retirement finished its pass without waiting for the lock", credentialId).ConfigureAwait(false); + + (await LoadCredentialAsync(credentialId).ConfigureAwait(false)).ShouldSatisfyAllConditions(c => c.EncryptedPayload.ShouldNotBeEmpty("somebody is writing the token right now, and a retirement under it would blank what that write lands"), c => c.DeletedDate.ShouldBeNull(), c => c.Status.ShouldBe(CredentialStatus.Active)); + (await LoadRepositoryAsync(repositoryId).ConfigureAwait(false)).Status.ShouldBe(RepositoryStatus.Active); + } + finally + { + await write.DisposeAsync().ConfigureAwait(false); + } + + await RetireThroughAsync(interceptor: null).ConfigureAwait(false); + + (await LoadCredentialAsync(credentialId).ConfigureAwait(false)).ShouldSatisfyAllConditions(c => c.EncryptedPayload.ShouldBeEmpty("a token that was busy is retired once nothing holds it, however many passes it was skipped: skipping it was not a failure"), c => c.DeletedDate.ShouldNotBeNull(), c => c.Status.ShouldBe(CredentialStatus.Revoked)); + (await LoadRepositoryAsync(repositoryId).ConfigureAwait(false)).Status.ShouldBe(RepositoryStatus.Error); + await AssertLockFreeAsync(credentialId).ConfigureAwait(false); + } + + [Fact] + public async Task A_re_authorization_that_arrives_while_a_retirement_is_under_way_is_refused_and_not_reported_done_and_then_undone() + { + using var gitlab = new GitLabDoorkeeperStub(); + var world = await SeedWorldAsync(gitlab).ConfigureAwait(false); + var (credentialId, repositoryId, personId) = await SeedDepartedMembersTokenWithOwnerAsync(world).ConfigureAwait(false); + gitlab.SeedPersonalAccessToken("departed", "glpat-fresh"); + Task? replace = null; + var finishedBeforeTheRetirement = false; + var arrives = new AfterReadOf(ReadsRepositories, async () => + { + await JoinTeamAsync(world, personId).ConfigureAwait(false); + replace = Task.Run(() => ReplaceTokenAsync(world, personId, credentialId, "glpat-fresh")); + await Task.Delay(LongEnoughToHaveFinished).ConfigureAwait(false); + finishedBeforeTheRetirement = replace.IsCompleted; + }); + + await RetireThroughAsync(arrives).ConfigureAwait(false); + + arrives.Ran.ShouldBeTrue("fixture check: the person rejoined and re-authorized after the sweep had judged the token departed and before it saved"); + finishedBeforeTheRetirement.ShouldBeFalse("the retirement holds the token's lock while it saves, so the re-authorization cannot finish inside it"); + await Should.ThrowAsync(() => WithinPatienceAsync(replace!, "the re-authorization that waited for the retirement finished", credentialId)).ConfigureAwait(false); + + (await LoadCredentialAsync(credentialId).ConfigureAwait(false)).ShouldSatisfyAllConditions(c => c.EncryptedPayload.ShouldBeEmpty("the retirement stands, and the person was told the re-authorization did not take — not told it did, and then had it blanked"), c => c.DeletedDate.ShouldNotBeNull()); + (await LoadRepositoryAsync(repositoryId).ConfigureAwait(false)).Status.ShouldBe(RepositoryStatus.Error); + } + + [Fact] + public async Task A_member_who_rejoined_and_re_authorized_before_the_sweep_took_their_token_keeps_it() + { + using var gitlab = new GitLabDoorkeeperStub(); + var world = await SeedWorldAsync(gitlab).ConfigureAwait(false); + var (credentialId, repositoryId, personId) = await SeedDepartedMembersTokenWithOwnerAsync(world).ConfigureAwait(false); + gitlab.SeedPersonalAccessToken("departed", "glpat-fresh"); + var rejoins = new AfterReadOf(ReadsDepartedCandidates, async () => + { + await JoinTeamAsync(world, personId).ConfigureAwait(false); + await ReplaceTokenAsync(world, personId, credentialId, "glpat-fresh").ConfigureAwait(false); + }); + + await RetireThroughAsync(rejoins).ConfigureAwait(false); + + rejoins.Ran.ShouldBeTrue("fixture check: the person rejoined and re-authorized after the sweep had read its candidates"); + (await TokenOfAsync(credentialId).ConfigureAwait(false)).ShouldBe("glpat-fresh", "the sweep judged the token again under its lock, found its person standing in the team, and left the new token alone"); + (await LoadCredentialAsync(credentialId).ConfigureAwait(false)).DeletedDate.ShouldBeNull(); + (await LoadRepositoryAsync(repositoryId).ConfigureAwait(false)).Status.ShouldBe(RepositoryStatus.Active); + } + + // ─── Plumbing ────────────────────────────────────────────────────────────────────────────────────────── + + private sealed record Stranded(Guid Dead, Guid Successor, Guid Repository, Guid Hook); + + /// + /// What a sweep repairs: a dead credential of Nacho's with a repository and a group hook still bound to it, and a live one + /// of his connected after it to follow — the dead one at the front of the queue, so every pass reaches it first. + /// + private async Task SeedStrandedAsync(World world, DeadAs deadAs) + { + var bound = await SeedBoundDeadCredentialAsync(world, deadAs).ConfigureAwait(false); + var successor = await SeedOAuthCredentialAsync(world, world.NachoId, "at-1", "rt-1").ConfigureAwait(false); + + return new Stranded(bound.Dead, successor, bound.Repository, bound.Hook); + } + + /// The same, with nobody connected after it yet: what a plain connect of Nacho's is about to carry forward. + private async Task SeedBoundRefusedCredentialAsync(World world) => await SeedBoundDeadCredentialAsync(world, DeadAs.Refused).ConfigureAwait(false); + + private async Task SeedBoundDeadCredentialAsync(World world, DeadAs deadAs) + { + world.GitLab.Seed(GitLabDoorkeeperStub.DefaultClientId, "nacho", "at-0", "rt-0"); + var disconnected = deadAs == DeadAs.Disconnected; + + var dead = await SeedOAuthCredentialAsync(world, world.NachoId, "at-0", "rt-0", disconnected: disconnected, status: disconnected ? CredentialStatus.Active : CredentialStatus.Error, lastError: disconnected ? null : "needs reconnect: invalid_grant", id: FrontOfTheQueue()).ConfigureAwait(false); + await SeedIdentityAsync(world, world.NachoId, dead, "nacho", deleted: disconnected).ConfigureAwait(false); + var repository = await SeedRepositoryAsync(world, dead, disconnected ? RepositoryStatus.Error : RepositoryStatus.Active).ConfigureAwait(false); + var hook = await SeedHookAsync(world, dead, RepositoryWebhookRegistrationStatus.DeadLettered).ConfigureAwait(false); + + return new Stranded(dead, Guid.Empty, repository, hook); + } + + /// A live personal token of someone who never joined the team — what the retirement sweep is for — with a repository bound through it. + private async Task<(Guid Credential, Guid Repository)> SeedDepartedMembersTokenAsync(World world) + { + var (credentialId, repositoryId, _) = await SeedDepartedMembersTokenWithOwnerAsync(world).ConfigureAwait(false); + + return (credentialId, repositoryId); + } + + private async Task<(Guid Credential, Guid Repository, Guid Person)> SeedDepartedMembersTokenWithOwnerAsync(World world) + { + var person = await SeedUserAsync().ConfigureAwait(false); + var credentialId = await SeedTokenCredentialAsync(world, person, "glpat-departed", id: FrontOfTheQueue()).ConfigureAwait(false); + var repositoryId = await SeedRepositoryAsync(world, credentialId, RepositoryStatus.Active).ConfigureAwait(false); + + return (credentialId, repositoryId, person); + } + + private async Task JoinTeamAsync(World world, Guid userId) + { + using var scope = _fixture.BeginScope(); + var db = scope.Resolve(); + + db.TeamMembership.Add(Membership(world.TeamId, userId, TeamRole.Member)); + await db.SaveChangesAsync().ConfigureAwait(false); + } + + /// An id that sorts ahead of any random one: a sweep reads its candidates in id order, so a candidate here is in the first page whatever else the shared database holds. + private static Guid FrontOfTheQueue() => Guid.Parse($"00000000-0000-0000-0000-{Guid.NewGuid().ToString("N")[..12]}"); + + private async Task SweepThroughAsync(IInterceptor? interceptor) + { + using var scope = ScopeThrough(interceptor); + + await scope.Resolve().Send(new RepairDisconnectedCredentialBindingsCommand { BatchSize = 50 }).ConfigureAwait(false); + } + + private async Task RetireThroughAsync(IInterceptor? interceptor) + { + using var scope = ScopeThrough(interceptor); + + await scope.Resolve().Send(new RetireDepartedMembersCredentialsCommand { BatchSize = 50 }).ConfigureAwait(false); + } + + private ILifetimeScope ScopeThrough(IInterceptor? interceptor) => interceptor == null ? _fixture.BeginScope() : _fixture.BeginScope(Through(interceptor)); + + /// The scope's database commands pass : production's options with one more interceptor on them. + private Action Through(IInterceptor interceptor) => builder => + { + DbContextOptions production; + + using (var probe = _fixture.BeginScope()) + production = probe.Resolve>(); + + builder.RegisterInstance(new DbContextOptionsBuilder(production).AddInterceptors(interceptor).Options).As>().SingleInstance(); + }; + + /// What a write of the credential's token holds while it is under way: the in-process semaphore and the advisory lock, or — as another pod's — the advisory lock alone. + private async Task HoldLockOfAsync(Guid credentialId, HeldBy heldBy) + { + using var scope = _fixture.BeginScope(); + var crossProcess = scope.Resolve(); + + return heldBy == HeldBy.ThisProcess + ? await CredentialRefreshLock.AcquireAsync(crossProcess, credentialId, CancellationToken.None).ConfigureAwait(false) + : await crossProcess.AcquireAsync(CredentialRefreshLock.KeyOf(credentialId), CancellationToken.None).ConfigureAwait(false); + } + + /// What the hourly health probe writes when the provider accepts a credential it had refused: the credential back to Active, on a connection of its own and without the refresh lock. False when the row no longer holds the payload it judged. + private async Task AcceptAgainAsync(Guid credentialId) + { + var judged = (await LoadCredentialAsync(credentialId).ConfigureAwait(false)).EncryptedPayload; + + using var scope = _fixture.BeginScope(); + + return await scope.Resolve().MarkValidatedAsync(credentialId, judged, DateTimeOffset.UtcNow, expiry: null, CancellationToken.None).ConfigureAwait(false); + } + + /// The lock was given back: both tiers are taken at once. A lock a sweep leaked would hang every later write of the credential, so this waits only as long as a free lock takes. + private async Task AssertLockFreeAsync(Guid credentialId) + { + using var scope = _fixture.BeginScope(); + + await using var held = await CredentialRefreshLock.AcquireAsync(scope.Resolve(), credentialId, CancellationToken.None).WaitAsync(TimeSpan.FromSeconds(5)).ConfigureAwait(false); + } + + private async Task AssertNothingBoundAsync(Guid credentialId) + { + using var scope = _fixture.BeginScope(); + var db = scope.Resolve(); + + (await db.Repository.AsNoTracking().CountAsync(r => r.CredentialId == credentialId && r.DeletedDate == null).ConfigureAwait(false)).ShouldBe(0, "nothing was carried onto the other credential"); + (await db.ConnectionWebhook.AsNoTracking().CountAsync(h => h.CredentialId == credentialId).ConfigureAwait(false)).ShouldBe(0); + } + + // The statements below are the ones that read what a verdict is taken on. They are matched by the tables and clauses that + // make them what they are, not by position, so a reordering of the service does not move the gap the test makes. + + private static bool ReadsStrandedCandidates(string sql) => Reads(sql, "FROM credential AS", "FROM repository AS", "ORDER BY"); + + private static bool ReadsOwnersCredentials(string sql) => Reads(sql, "FROM credential AS", "owner_user_id", "<>"); + + private static bool ReadsRepositories(string sql) => Reads(sql, "FROM repository AS"); + + private static bool ReadsDepartedCandidates(string sql) => Reads(sql, "FROM credential AS", "team_membership"); + + private static bool Reads(string sql, params string[] fragments) => sql.StartsWith("SELECT", StringComparison.Ordinal) && fragments.All(f => sql.Contains(f, StringComparison.Ordinal)); + + /// Runs then once, right after the nth read whose SQL matches — a request landing in the gap that follows the statement that read the candidate. + private sealed class AfterReadOf : DbCommandInterceptor + { + private readonly Func _matches; + private readonly Func _then; + private readonly int _nth; + private int _seen; + private int _ran; + + public AfterReadOf(Func matches, Func then, int nth = 1) + { + _matches = matches; + _then = then; + _nth = nth; + } + + public bool Ran => Volatile.Read(ref _ran) == 1; + + public override async ValueTask ReaderExecutedAsync(DbCommand command, CommandExecutedEventData eventData, DbDataReader result, CancellationToken cancellationToken = default) + { + if (!_matches(command.CommandText) || Interlocked.Increment(ref _seen) != _nth) return result; + + await _then().ConfigureAwait(false); + Volatile.Write(ref _ran, 1); + + return result; + } + } +} diff --git a/backend/tests/CodeSpace.IntegrationTests/Credentials/CredentialReauthorizationFlowTests.cs b/backend/tests/CodeSpace.IntegrationTests/Credentials/CredentialReauthorizationFlowTests.cs index cff8b973c..a667a6f2a 100644 --- a/backend/tests/CodeSpace.IntegrationTests/Credentials/CredentialReauthorizationFlowTests.cs +++ b/backend/tests/CodeSpace.IntegrationTests/Credentials/CredentialReauthorizationFlowTests.cs @@ -823,32 +823,32 @@ private async Task SeedInstanceAtAsync(World world, string baseUrl, string /// Revoked, the payload blanked, the row still listed; is what a departure or a connection /// delete leaves on top of that — soft-deleted. /// - private async Task SeedOAuthCredentialAsync(World world, Guid? ownerId, string accessToken, string refreshToken, bool disconnected = false, bool retired = false, bool expired = false, CredentialStatus status = CredentialStatus.Active, string? lastError = null, CredentialOwnership ownership = CredentialOwnership.Personal, Guid? instanceId = null) + private async Task SeedOAuthCredentialAsync(World world, Guid? ownerId, string accessToken, string refreshToken, bool disconnected = false, bool retired = false, bool expired = false, CredentialStatus status = CredentialStatus.Active, string? lastError = null, CredentialOwnership ownership = CredentialOwnership.Personal, Guid? instanceId = null, Guid? id = null) { using var scope = _fixture.BeginScope(); var payload = new OAuthPayload { AccessToken = accessToken, RefreshToken = refreshToken, ExpiresAt = expired ? DateTimeOffset.UtcNow.AddMinutes(-1) : DateTimeOffset.UtcNow.AddHours(2) }; var sealedPayload = scope.Resolve().Encrypt(scope.Resolve().Serialize(payload)); - return await InsertCredentialAsync(scope, world, ownerId, AuthType.OAuth, sealedPayload, payload.ExpiresAt, disconnected || retired, retired, status, lastError, ownership, instanceId).ConfigureAwait(false); + return await InsertCredentialAsync(scope, world, ownerId, AuthType.OAuth, sealedPayload, payload.ExpiresAt, disconnected || retired, retired, status, lastError, ownership, instanceId, id).ConfigureAwait(false); } - private async Task SeedTokenCredentialAsync(World world, Guid? ownerId, string token, AuthType authType = AuthType.Pat, CredentialOwnership ownership = CredentialOwnership.Personal, CredentialStatus status = CredentialStatus.Active, string? lastError = null, bool retired = false, DateTimeOffset? expires = null, Guid? instanceId = null) + private async Task SeedTokenCredentialAsync(World world, Guid? ownerId, string token, AuthType authType = AuthType.Pat, CredentialOwnership ownership = CredentialOwnership.Personal, CredentialStatus status = CredentialStatus.Active, string? lastError = null, bool retired = false, DateTimeOffset? expires = null, Guid? instanceId = null, Guid? id = null) { using var scope = _fixture.BeginScope(); CredentialPayload payload = authType == AuthType.GroupAccessToken ? new GroupAccessTokenPayload { Token = token } : new PatPayload { Token = token }; var sealedPayload = scope.Resolve().Encrypt(scope.Resolve().Serialize(payload)); - return await InsertCredentialAsync(scope, world, ownerId, authType, retired ? string.Empty : sealedPayload, expires, retired, retired, status, lastError, ownership, instanceId).ConfigureAwait(false); + return await InsertCredentialAsync(scope, world, ownerId, authType, retired ? string.Empty : sealedPayload, expires, retired, retired, status, lastError, ownership, instanceId, id).ConfigureAwait(false); } - private static async Task InsertCredentialAsync(ILifetimeScope scope, World world, Guid? ownerId, AuthType authType, string encryptedPayload, DateTimeOffset? expires, bool blank, bool retired, CredentialStatus status, string? lastError, CredentialOwnership ownership, Guid? instanceId) + private static async Task InsertCredentialAsync(ILifetimeScope scope, World world, Guid? ownerId, AuthType authType, string encryptedPayload, DateTimeOffset? expires, bool blank, bool retired, CredentialStatus status, string? lastError, CredentialOwnership ownership, Guid? instanceId, Guid? id) { var db = scope.Resolve(); var actor = ownerId ?? world.BossId; var credential = new Credential { - Id = Guid.NewGuid(), + Id = id ?? Guid.NewGuid(), TeamId = world.TeamId, ProviderInstanceId = instanceId ?? world.InstanceId, OwnerUserId = ownerId, diff --git a/backend/tests/CodeSpace.IntegrationTests/OAuth/RepairSweepFairnessFlowTests.cs b/backend/tests/CodeSpace.IntegrationTests/OAuth/RepairSweepFairnessFlowTests.cs index 541606eb4..175970c9e 100644 --- a/backend/tests/CodeSpace.IntegrationTests/OAuth/RepairSweepFairnessFlowTests.cs +++ b/backend/tests/CodeSpace.IntegrationTests/OAuth/RepairSweepFairnessFlowTests.cs @@ -2,6 +2,7 @@ using CodeSpace.Core.Persistence.Db; using CodeSpace.Core.Persistence.Entities; using CodeSpace.Core.Services.Credentials; +using CodeSpace.Core.Services.OAuth; using CodeSpace.IntegrationTests.Infrastructure; using CodeSpace.Messages.Commands.Credentials; using CodeSpace.Messages.Enums; @@ -56,7 +57,10 @@ public async Task Credentials_that_can_never_be_repaired_do_not_starve_the_ones_ passes.ShouldBeGreaterThanOrEqualTo(1); foreach (var stuck in poisoned) + { (await _kit.LoadRepositoryAsync(stuck.Repository)).CredentialId.ShouldBe(stuck.Dead, "a repair that failed leaves its binding exactly where it was — nothing half-moved"); + await AssertLockFreeAsync(stuck.Dead); + } } [Fact] @@ -102,6 +106,14 @@ public async Task A_candidate_that_has_failed_several_passes_running_is_left_alo (await _kit.LoadRepositoryAsync(permanent.Repository)).CredentialId.ShouldBe(permanent.Dead, "failing that many passes in a row put it out of the sweep's rotation, so a pass within the window does not attempt it even though it would now succeed"); } + /// A candidate that failed gave its lock back: a re-authorization of it is not left waiting behind a sweep that is long gone. + private async Task AssertLockFreeAsync(Guid credentialId) + { + using var scope = _fixture.BeginScope(); + + await using var held = await CredentialRefreshLock.AcquireAsync(scope.Resolve(), credentialId, CancellationToken.None).WaitAsync(TimeSpan.FromSeconds(5)); + } + private async Task SweepOnceAsync() { using var scope = _fixture.BeginScope(); diff --git a/backend/tests/CodeSpace.IntegrationTests/OAuth/RetireSweepFairnessFlowTests.cs b/backend/tests/CodeSpace.IntegrationTests/OAuth/RetireSweepFairnessFlowTests.cs index 8de7c7a48..7be6ec85c 100644 --- a/backend/tests/CodeSpace.IntegrationTests/OAuth/RetireSweepFairnessFlowTests.cs +++ b/backend/tests/CodeSpace.IntegrationTests/OAuth/RetireSweepFairnessFlowTests.cs @@ -2,6 +2,7 @@ using CodeSpace.Core.Persistence.Db; using CodeSpace.Core.Persistence.Entities; using CodeSpace.Core.Services.Credentials; +using CodeSpace.Core.Services.OAuth; using CodeSpace.IntegrationTests.Infrastructure; using CodeSpace.Messages.Commands.Credentials; using CodeSpace.Messages.Enums; @@ -63,6 +64,7 @@ public async Task Tokens_that_can_never_be_retired_do_not_starve_the_departed_to credential.DeletedDate.ShouldBeNull("a retirement that failed leaves the token exactly as it was — nothing half-done"); credential.Status.ShouldBe(CredentialStatus.Active); + await AssertLockFreeAsync(stuck); } } @@ -92,6 +94,14 @@ public async Task A_token_that_has_failed_several_passes_running_is_left_alone_f (await _kit.LoadCredentialAsync(permanent)).DeletedDate.ShouldBeNull("failing that many passes in a row put it out of the sweep's rotation, so a pass within the window does not attempt it even though it would now succeed"); } + /// A candidate that failed gave its lock back: a re-authorization of it is not left waiting behind a sweep that is long gone. + private async Task AssertLockFreeAsync(Guid credentialId) + { + using var scope = _fixture.BeginScope(); + + await using var held = await CredentialRefreshLock.AcquireAsync(scope.Resolve(), credentialId, CancellationToken.None).WaitAsync(TimeSpan.FromSeconds(5)); + } + private async Task SweepOnceAsync() { using var scope = _fixture.BeginScope(); diff --git a/backend/tests/CodeSpace.UnitTests/Architecture/ScopedTransactionInventoryTests.cs b/backend/tests/CodeSpace.UnitTests/Architecture/ScopedTransactionInventoryTests.cs index 88ff62660..727539610 100644 --- a/backend/tests/CodeSpace.UnitTests/Architecture/ScopedTransactionInventoryTests.cs +++ b/backend/tests/CodeSpace.UnitTests/Architecture/ScopedTransactionInventoryTests.cs @@ -65,6 +65,7 @@ public class ScopedTransactionInventoryTests ["Agents/Mcp/IToolCallLedgerService.cs"] = (1, "rolls back and CONTINUES on a lost claim race, then re-reads the winner; a joined rollback would take the caller's transaction with it."), ["Agents/Publish/ArtifactManifestStore.cs"] = (1, "already joins, but through a SAVEPOINT: a failed pointer replacement must discard only its own writes, which a plain join cannot do."), ["Agents/Publish/IPublishManifestStore.cs"] = (1, "rolls back and CONTINUES into the fenced-update fallback when the first insert loses its race."), + ["Credentials/CredentialSuccessionService.cs"] = (1, "the repair sweep's per-credential unit, inside a sweep that has no transaction to join: the dead credential's row is held from its verdict to this commit and must be released by it — joined, one transaction over the batch would keep every credential's row until the last was repaired and let one failing credential undo the others' repairs."), ["Learning/LessonQualifier.cs"] = (1, "its advisory lock is a GLOBAL key; joined, one long command would serialize every qualification sweep in the deployment behind its tail."), ["Supervisor/Executors/RealSupervisorActionExecutor.Spawn.cs"] = (1, "the authorization wave must roll back to zero residue when the supervisor catches the fault and carries on; joined, that discard would be a no-op and replay would see a partial wave."), ["Supervisor/Observation/SupervisorDecisionObservationMetadataReader.cs"] = (1, "already joins; the RepeatableRead snapshot it opens is only meaningful for the transaction it owns."), diff --git a/backend/tests/CodeSpace.UnitTests/OAuth/CredentialRefreshLockTests.cs b/backend/tests/CodeSpace.UnitTests/OAuth/CredentialRefreshLockTests.cs index ab1986021..78c3625b4 100644 --- a/backend/tests/CodeSpace.UnitTests/OAuth/CredentialRefreshLockTests.cs +++ b/backend/tests/CodeSpace.UnitTests/OAuth/CredentialRefreshLockTests.cs @@ -6,7 +6,8 @@ namespace CodeSpace.UnitTests.OAuth; /// /// 🟢 High fidelity — the production and its real in-process semaphores; only the /// cross-process tier is a recording fake, because ordering against it is the claim. A refresh and a re-authorization both -/// take this one lock, and what they share is process-wide state, so these pin the properties that make "the same lock" +/// take this one lock — and the sweeps that move or retire what hangs on a dead credential take it without waiting +/// (TryAcquireAsync) — and what they share is process-wide state, so these pin the properties that make "the same lock" /// true: the same credential excludes, another does not, and the two tiers are taken and given back in one fixed order. /// [Trait("Category", "Unit")] @@ -112,6 +113,83 @@ public async Task Disposing_a_handle_twice_releases_once() third.IsCompleted.ShouldBeFalse(); } + [Fact] + public async Task A_try_acquire_of_a_free_credential_takes_both_tiers_and_gives_them_back() + { + var id = Guid.NewGuid(); + var tryLock = new RecordingTryLock(); + + var held = await CredentialRefreshLock.TryAcquireAsync(tryLock, id, CancellationToken.None); + + held.ShouldNotBeNull("nothing holds the credential"); + tryLock.Events.ShouldBe(new[] { "acquire" }); + + await held.DisposeAsync(); + + tryLock.Events.ShouldBe(new[] { "acquire", "release" }); + await using var again = await CredentialRefreshLock.AcquireAsync(new RecordingCrossProcessLock(), id, CancellationToken.None).WaitAsync(Patience); + } + + [Fact] + public async Task A_try_acquire_of_a_credential_this_process_holds_returns_at_once_without_asking_the_advisory_lock() + { + var id = Guid.NewGuid(); + var tryLock = new RecordingTryLock(); + var holder = await CredentialRefreshLock.AcquireAsync(new RecordingCrossProcessLock(), id, CancellationToken.None); + + var attempt = await CredentialRefreshLock.TryAcquireAsync(tryLock, id, CancellationToken.None).WaitAsync(Patience); + + attempt.ShouldBeNull("a refresh or re-authorization in this process holds the credential, and a sweep does not queue behind it"); + tryLock.Events.ShouldBeEmpty("the in-process semaphore is the first tier: nothing is asked of Postgres once it says busy"); + + await holder.DisposeAsync(); + + await using var afterwards = await CredentialRefreshLock.TryAcquireAsync(tryLock, id, CancellationToken.None); + afterwards.ShouldNotBeNull("once the holder lets go the credential is free for the sweep again"); + } + + [Fact] + public async Task A_try_acquire_of_a_credential_another_pod_holds_returns_null_and_gives_the_semaphore_back() + { + var id = Guid.NewGuid(); + var tryLock = new RecordingTryLock { Busy = true }; + + var attempt = await CredentialRefreshLock.TryAcquireAsync(tryLock, id, CancellationToken.None); + + attempt.ShouldBeNull("another pod holds the advisory lock"); + tryLock.Events.ShouldBe(new[] { "refused" }); + await using var again = await CredentialRefreshLock.AcquireAsync(new RecordingCrossProcessLock(), id, CancellationToken.None).WaitAsync(Patience); + } + + [Fact] + public async Task A_try_acquire_whose_advisory_attempt_fails_gives_the_semaphore_back() + { + var id = Guid.NewGuid(); + + var failure = await Should.ThrowAsync(async () => await CredentialRefreshLock.TryAcquireAsync(new RecordingTryLock { Fail = true }, id, CancellationToken.None)); + + failure.Message.ShouldContain("advisory"); + await using var again = await CredentialRefreshLock.AcquireAsync(new RecordingCrossProcessLock(), id, CancellationToken.None).WaitAsync(Patience); + } + + [Fact] + public async Task A_credential_held_by_a_try_acquire_holds_up_a_waiting_acquire_until_it_is_disposed() + { + // One semaphore for both ways in: a sweep that holds a credential excludes a re-authorization exactly as a refresh would. + var id = Guid.NewGuid(); + var crossProcess = new RecordingCrossProcessLock(); + var sweep = await CredentialRefreshLock.TryAcquireAsync(new RecordingTryLock(), id, CancellationToken.None); + + var reauthorization = CredentialRefreshLock.AcquireAsync(crossProcess, id, CancellationToken.None); + + await Task.Delay(ShortWait); + reauthorization.IsCompleted.ShouldBeFalse("the re-authorization waits for the sweep's hold, so it cannot write while the sweep is moving"); + + await sweep!.DisposeAsync(); + + await using var handle = await reauthorization.WaitAsync(Patience); + } + private sealed class RecordingCrossProcessLock : ICrossProcessLock { private readonly object _gate = new(); @@ -144,4 +222,39 @@ public ValueTask DisposeAsync() } } } + + private sealed class RecordingTryLock : ICrossProcessTryLock + { + private readonly object _gate = new(); + private readonly List _events = new(); + + public bool Busy { get; init; } + + public bool Fail { get; init; } + + public List Events { get { lock (_gate) { return _events.ToList(); } } } + + public Task TryAcquireAsync(long key, CancellationToken cancellationToken) + { + if (Fail) throw new InvalidOperationException("simulated: the advisory lock could not be tried"); + + lock (_gate) { _events.Add(Busy ? "refused" : "acquire"); } + + return Task.FromResult(Busy ? null : new Release(this)); + } + + private sealed class Release : IAsyncDisposable + { + private readonly RecordingTryLock _owner; + + public Release(RecordingTryLock owner) { _owner = owner; } + + public ValueTask DisposeAsync() + { + lock (_owner._gate) { _owner._events.Add("release"); } + + return ValueTask.CompletedTask; + } + } + } }