37 lines
1.3 KiB
C#
37 lines
1.3 KiB
C#
using System.Collections.Concurrent;
|
|
|
|
namespace w4c_workflows.Services;
|
|
|
|
/// <summary>
|
|
/// Serializes concurrent workflow syncs for the same <c>(tenant, repo)</c> within
|
|
/// one process. Two overlapping <c>POST /sync</c> calls would otherwise both
|
|
/// insert the same deterministic workflow/task ids and collide on the primary key
|
|
/// (<c>DbUpdateException</c>). The gate makes the second call wait, so it observes
|
|
/// the first call's committed rows and upserts instead of duplicating.
|
|
///
|
|
/// For multi-replica deployments this is combined with the distributed
|
|
/// <see cref="ILeaseService"/> lease in <see cref="WorkflowSyncService"/>.
|
|
/// </summary>
|
|
public sealed class SyncGate
|
|
{
|
|
private readonly ConcurrentDictionary<string, SemaphoreSlim> _gates = new(StringComparer.Ordinal);
|
|
|
|
public async Task<IDisposable> AcquireAsync(string tenantId, string repoName, CancellationToken ct)
|
|
{
|
|
var gate = _gates.GetOrAdd(tenantId + "|" + repoName, _ => new SemaphoreSlim(1, 1));
|
|
await gate.WaitAsync(ct);
|
|
return new Releaser(gate);
|
|
}
|
|
|
|
private sealed class Releaser(SemaphoreSlim gate) : IDisposable
|
|
{
|
|
private int _released;
|
|
|
|
public void Dispose()
|
|
{
|
|
if (Interlocked.Exchange(ref _released, 1) == 0)
|
|
gate.Release();
|
|
}
|
|
}
|
|
}
|