fix: hold lock on reddit id during write operations

These changes will serialize operations against the specific reddit post id in-process. There was a race condition where the background processor could pull in a reddit post id for processing, read all of the links, make the reddit API calls, and then remove the link from the queue. Specifically, the race condition is between the reddit API calls and removing the link from the queue (making it as processed).

While the database operations are (under normal circumstances) extremely fast, the reddit API operations have consistently profiled as relatively slow (averaging 250ms per call), and each link can require up to 4 API calls to be made. This leaves a non-deterministic window where additional links sent to the API will never be posted.

Author: kyleratti

ApplicationData/ApplicationData.csproj

@@ -24,6 +24,7 @@
       </PackageReference>
       <PackageReference Include="Microsoft.Data.Sqlite" Version="8.0.2" />
       <PackageReference Include="Microsoft.Extensions.Caching.Abstractions" Version="8.0.0" />
+      <PackageReference Include="Nito.AsyncEx" Version="5.1.2" />
       <PackageReference Include="SnooBrowser" Version="3.1.1" />
       <PackageReference Include="System.Linq.Async" Version="6.0.1" />
     </ItemGroup>

ApplicationData/Services/ResourceAccessManager.cs

@@ -0,0 +1,51 @@
+using Microsoft.Extensions.Caching.Memory;
+using Nito.AsyncEx;
+
+namespace ApplicationData.Services;
+
+public sealed class ResourceAccessManager(IMemoryCache _memoryCache)
+{
+	private readonly AsyncLock _resourceWriteLock = new();
+
+	public async Task<IDisposable> ObtainExclusiveAccess(string resourceId, CancellationToken cancellationToken)
+	{
+		var resourceLock = await GetOrCreateResourceLock(resourceId, cancellationToken);
+		return await resourceLock.LockAsync(cancellationToken);
+	}
+
+	public async Task<T> WithExclusiveAccess<T>(string resourceId, Func<Task<T>> action, CancellationToken cancellationToken)
+	{
+		var resourceLock = await GetOrCreateResourceLock(resourceId, cancellationToken);
+		using var _ = await resourceLock.LockAsync(cancellationToken);
+
+		return await action();
+	}
+
+	public async Task WithExclusiveAccess(string resourceId, Func<Task> action, CancellationToken cancellationToken) =>
+		await WithExclusiveAccess(
+			resourceId,
+			async () =>
+			{
+				await action();
+				return 0;
+			},
+			cancellationToken);
+
+	private async Task<AsyncLock> GetOrCreateResourceLock(string resourceId, CancellationToken cancellationToken)
+	{
+		using var _ = await _resourceWriteLock.LockAsync(cancellationToken);
+
+		var resourceLock = _memoryCache.GetOrCreate(
+			resourceId,
+			entry =>
+			{
+				entry.SetSlidingExpiration(TimeSpan.FromMinutes(10));
+				return new AsyncLock();
+			});
+
+		if (resourceLock is null)
+			throw new InvalidOperationException($"Resource lock was not created or retrieved (ResourceId={resourceId}");
+
+		return resourceLock;
+	}
+}

BackgroundProcessor/BackgroundProcessor.csproj

@@ -7,6 +7,7 @@
         <UserSecretsId>dotnet-BackgroundProcessor-35541A68-9074-4C0A-9003-D5B4779CC1E8</UserSecretsId>
         <DockerDefaultTargetOS>Linux</DockerDefaultTargetOS>
         <LangVersion>12</LangVersion>
+        <OutputType>Library</OutputType>
     </PropertyGroup>
 
     <ItemGroup>

BackgroundProcessor/Processors/LinkProcessor.cs

@@ -12,35 +12,17 @@ namespace BackgroundProcessor.Processors;
 
 // ReSharper disable once UnusedType.Global
 // it is auto-created
-public class LinkProcessor : IBackgroundProcessor
+public class LinkProcessor(
+	ILogger<LinkProcessor> _logger,
+	LinkProvider _linkProvider,
+	RedditCommentProvider _commentProvider,
+	CommentBrowser _commentBrowser,
+	UserProvider _userProvider,
+	TemplateCache _templateCache,
+	IDbConnectionFactory _dbConnectionFactory,
+	ResourceAccessManager _resourceAccessManager
+) : IBackgroundProcessor
 {
-	private readonly ILogger<LinkProcessor> _logger;
-	private readonly LinkProvider _linkProvider;
-	private readonly RedditCommentProvider _commentProvider;
-	private readonly CommentBrowser _commentBrowser;
-	private readonly UserProvider _userProvider;
-	private readonly TemplateCache _templateCache;
-	private readonly IDbConnectionFactory _dbConnectionFactory;
-
-	public LinkProcessor(
-		ILogger<LinkProcessor> logger,
-		LinkProvider linkProvider,
-		RedditCommentProvider commentProvider,
-		CommentBrowser commentBrowser,
-		UserProvider userProvider,
-		TemplateCache templateCache,
-		IDbConnectionFactory dbConnectionFactory
-	)
-	{
-		_logger = logger;
-		_linkProvider = linkProvider;
-		_commentProvider = commentProvider;
-		_commentBrowser = commentBrowser;
-		_userProvider = userProvider;
-		_templateCache = templateCache;
-		_dbConnectionFactory = dbConnectionFactory;
-	}
-
 	/// <inheritdoc />
 	public async Task Process(CancellationToken cancellationToken)
 	{
@@ -50,6 +32,8 @@ public async Task Process(CancellationToken cancellationToken)
 
 		foreach (var item in postIdsWithPendingChanges)
 		{
+			using var redditPostIdLock = await _resourceAccessManager.ObtainExclusiveAccess(item.RedditPostId, cancellationToken);
+
 			var links = await _linkProvider.GetLinksByRedditPostId(item.RedditPostId);
 			var maybeExistingComment = await _commentProvider.FindCommentIdByPostId(item.RedditPostId);
 

BackgroundProcessor/Program.cs

@@ -21,18 +21,21 @@ public class LinkController : ApiController
 {
 	private readonly LinkProvider _linkProvider;
 	private readonly SubmissionBrowser _submissionBrowser;
+	private readonly ResourceAccessManager _resourceAccessManager;
 
 	/// <summary>
 	/// C'tor
 	/// </summary>
 	public LinkController(
 		IHttpContextAccessor httpContextAccessor,
 		LinkProvider linkProvider,
-		SubmissionBrowser submissionBrowser
+		SubmissionBrowser submissionBrowser,
+		ResourceAccessManager resourceAccessManager
 	) : base(httpContextAccessor)
 	{
 		_linkProvider = linkProvider;
 		_submissionBrowser = submissionBrowser;
+		_resourceAccessManager = resourceAccessManager;
 	}
 
 	/// <summary>
@@ -47,7 +50,7 @@ SubmissionBrowser submissionBrowser
 	[SwaggerResponse((int)HttpStatusCode.Conflict,
 		description: "Conflict. This user has already provided this combination of reddit post ID, URL, and link type.",
 		typeof(LinkAlreadyExistsError))]
-	public async Task<IActionResult> SubmitLink([FromBody] SubmitLinkRequest linkRequest)
+	public async Task<IActionResult> SubmitLink([FromBody] SubmitLinkRequest linkRequest, CancellationToken cancellationToken)
 	{
 		var validUrl = GetValidUriOrFail(linkRequest.LinkUrl);
 		var linkKind = GetLinkKindOrFail(linkRequest.LinkType!.Value);
@@ -60,12 +63,16 @@ public async Task<IActionResult> SubmitLink([FromBody] SubmitLinkRequest linkReq
 				RedditPostId: linkRequest.RedditPostId!
 			));
 
-		var createResult = await _linkProvider.CreateLink(new NewLink(
-			redditPostId: linkRequest.RedditPostId,
-			linkUrl: validUrl.OriginalString,
-			linkKind,
-			ownerUserId: UserId
-		));
+		var createResult = await _resourceAccessManager.WithExclusiveAccess(
+			linkRequest.RedditPostId!,
+			async () =>
+				await _linkProvider.CreateLink(new NewLink(
+					redditPostId: linkRequest.RedditPostId,
+					linkUrl: validUrl.OriginalString,
+					linkKind,
+					ownerUserId: UserId
+				)),
+			cancellationToken);
 
 		if (createResult.IsFailure)
 		{
@@ -93,7 +100,7 @@ public async Task<IActionResult> SubmitLink([FromBody] SubmitLinkRequest linkReq
 	[SwaggerResponse((int)HttpStatusCode.NotFound,
 		description: "Not Found. This user does not have this combination of reddit post ID, URL, and link type.",
 		typeof(LinkNotFoundError))]
-	public async Task<IActionResult> DeleteLinkByLinkData([FromBody] DeleteLinkRequest linkRequest)
+	public async Task<IActionResult> DeleteLinkByLinkData([FromBody] DeleteLinkRequest linkRequest, CancellationToken cancellationToken)
 	{
 		var validUrl = GetValidUriOrFail(linkRequest.LinkUrl);
 		var linkKind = GetLinkKindOrFail(linkRequest.LinkType!.Value);
@@ -105,7 +112,10 @@ public async Task<IActionResult> DeleteLinkByLinkData([FromBody] DeleteLinkReque
 				Url: validUrl.OriginalString
 			));
 
-		await _linkProvider.DeleteLinkById(link.LinkId);
+		await _resourceAccessManager.WithExclusiveAccess(
+			linkRequest.RedditPostId,
+			() => _linkProvider.DeleteLinkById(link.LinkId),
+			cancellationToken);
 
 		return new OkObjectResult(new LinkDeleteQueuedSuccessfully(
 			Message: TranslatedStrings.LinkController.LinkDeleted,
@@ -125,15 +135,18 @@ public async Task<IActionResult> DeleteLinkByLinkData([FromBody] DeleteLinkReque
 	[SwaggerResponse((int)HttpStatusCode.NotFound,
 		description: "Not Found. This user does not have this combination of reddit post ID, URL, and link type.",
 		typeof(LinkIdNotFoundError))]
-	public async Task<IActionResult> DeleteLinkById([FromRoute] int linkId)
+	public async Task<IActionResult> DeleteLinkById([FromRoute] int linkId, CancellationToken cancellationToken)
 	{
 		if (!(await _linkProvider.FindLinkById(UserId, linkId)).Try(out var link))
 			return new NotFoundObjectResult(new LinkIdNotFoundError(
 				Message: TranslatedStrings.LinkController.LinkIdNotFound(linkId),
 				LinkId: linkId
 			));
 
-		await _linkProvider.DeleteLinkById(link.LinkId);
+		await _resourceAccessManager.WithExclusiveAccess(
+			link.RedditPostId,
+			() => _linkProvider.DeleteLinkById(link.LinkId),
+			cancellationToken);
 
 		return new OkObjectResult(new LinkDeleteQueuedSuccessfully(
 			Message: TranslatedStrings.LinkController.LinkDeleted,