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; + } + } + } }