From a938a7bcbb64096dbe1b41edd52ae059e87f3bb2 Mon Sep 17 00:00:00 2001 From: Hamza Alqurneh Date: Sun, 30 Aug 2026 12:45:46 +0300 Subject: [PATCH 1/2] fix: tell the cache about writes it was never told about MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Subscriptions, information types, notifiers, work groups, global values and bus gateways are held in memory for ten minutes, and a write announces itself with BroadcastRevoke so every instance drops its copy. Eleven write handlers never announced anything. Pausing was the worst of them: the receiving path reads PausedOn off the cached copy, so a paused integration kept taking messages, and resuming could find its own cached copy still paused and return without releasing what it held. Global values were doubly stale — no handler announced them, and Revoke() cleared five of the six keys Load() sets, so nothing could have cleared them anyway. Receive now, Aggregate now and ResetRetryUsage are left alone on purpose: their fields are read from the database, not the cache, so flushing everything on those would cost more than it buys. --- .../Resources/ApiGateways/AddPartner.cs | 7 +- SW.Bitween.Api/Resources/Documents/Create.cs | 8 +- SW.Bitween.Api/Resources/Documents/Delete.cs | 5 +- SW.Bitween.Api/Resources/Documents/Update.cs | 2 +- .../GlobalAdapterValuesSets/Create.cs | 5 +- .../GlobalAdapterValuesSets/Delete.cs | 5 +- .../GlobalAdapterValuesSets/Update.cs | 5 +- SW.Bitween.Api/Resources/Notifiers/Create.cs | 5 +- SW.Bitween.Api/Resources/Notifiers/Delete.cs | 5 +- SW.Bitween.Api/Resources/Notifiers/Update.cs | 5 +- .../Resources/Subscriptions/Delete.cs | 5 +- .../Resources/Subscriptions/Pause.cs | 9 +- .../Resources/Subscriptions/SaveMapper.cs | 2 +- SW.Bitween.Api/Resources/WorkGroups/Create.cs | 2 +- SW.Bitween.Api/Resources/WorkGroups/Delete.cs | 2 +- .../Services/Caching/InMemoryInfolinkCache.cs | 3 + .../Tests/CacheRevocationTests.cs | 134 ++++++++++++++++++ 17 files changed, 194 insertions(+), 15 deletions(-) create mode 100644 SW.Bitween.IntegrationTests/Tests/CacheRevocationTests.cs diff --git a/SW.Bitween.Api/Resources/ApiGateways/AddPartner.cs b/SW.Bitween.Api/Resources/ApiGateways/AddPartner.cs index 55edeed6..4606730b 100644 --- a/SW.Bitween.Api/Resources/ApiGateways/AddPartner.cs +++ b/SW.Bitween.Api/Resources/ApiGateways/AddPartner.cs @@ -15,13 +15,15 @@ public class AddPartner : ICommandHandler private readonly BitweenDbContext _dbContext; private readonly RequestContext _requestContext; private readonly AdapterRequirements _adapterRequirements; + private readonly IInfolinkCache _cache; public AddPartner(BitweenDbContext dbContext, RequestContext requestContext, - AdapterRequirements adapterRequirements) + AdapterRequirements adapterRequirements, IInfolinkCache cache) { _dbContext = dbContext; _requestContext = requestContext; _adapterRequirements = adapterRequirements; + _cache = cache; } public async Task Handle(int gatewayId, ApiGatewayPartnerCreate model) @@ -81,6 +83,9 @@ public async Task Handle(int gatewayId, ApiGatewayPartnerCreate model) _dbContext.Add(partnerLink); await _dbContext.SaveChangesAsync(); + // Attaching an existing integration changes nothing the cache holds, but staging a new + // one above creates a Subscription — and unconditional is what AddRoute does. + await _cache.BroadcastRevoke(); return null; } diff --git a/SW.Bitween.Api/Resources/Documents/Create.cs b/SW.Bitween.Api/Resources/Documents/Create.cs index 87b791f2..ba27ff41 100644 --- a/SW.Bitween.Api/Resources/Documents/Create.cs +++ b/SW.Bitween.Api/Resources/Documents/Create.cs @@ -17,12 +17,15 @@ public class Create : ICommandHandler private readonly BitweenDbContext _dbContext; private readonly RequestContext _requestContext; private readonly IBroadcast _broadcast; + private readonly IInfolinkCache _cache; - public Create(BitweenDbContext dbContext, RequestContext requestContext, IBroadcast broadcast) + public Create(BitweenDbContext dbContext, RequestContext requestContext, IBroadcast broadcast, + IInfolinkCache cache) { _dbContext = dbContext; _requestContext = requestContext; _broadcast = broadcast; + _cache = cache; } public async Task Handle(DocumentCreate model) @@ -78,6 +81,9 @@ public async Task Handle(DocumentCreate model) _dbContext.Add(trail); _dbContext.Add(entity); await _dbContext.SaveChangesAsync(); + // Routing resolves an information type by name off the cache, so a new one is + // invisible to it until this lands. + await _cache.BroadcastRevoke(); // A bus-enabled type adds a queue, and the consumer set is only rebuilt when asked. // Without this the queue is declared but nothing ever consumes it, until either an diff --git a/SW.Bitween.Api/Resources/Documents/Delete.cs b/SW.Bitween.Api/Resources/Documents/Delete.cs index cddf1571..b7edddb6 100644 --- a/SW.Bitween.Api/Resources/Documents/Delete.cs +++ b/SW.Bitween.Api/Resources/Documents/Delete.cs @@ -12,11 +12,13 @@ public class Delete : IDeleteHandler { private readonly BitweenDbContext _dbContext; private readonly RequestContext _requestContext; + private readonly IInfolinkCache _cache; - public Delete(BitweenDbContext dbContext, RequestContext requestContext) + public Delete(BitweenDbContext dbContext, RequestContext requestContext, IInfolinkCache cache) { _dbContext = dbContext; _requestContext = requestContext; + _cache = cache; } async public Task Handle(int key) @@ -24,6 +26,7 @@ async public Task Handle(int key) await _requestContext.EnsurePermission(_dbContext, Model.Permissions.Documents.Delete); await _dbContext.DeleteByKeyAsync(key); + await _cache.BroadcastRevoke(); return null; } } diff --git a/SW.Bitween.Api/Resources/Documents/Update.cs b/SW.Bitween.Api/Resources/Documents/Update.cs index 32f5f89c..21c5c618 100644 --- a/SW.Bitween.Api/Resources/Documents/Update.cs +++ b/SW.Bitween.Api/Resources/Documents/Update.cs @@ -102,7 +102,7 @@ public async Task Handle(int key, DocumentUpdate model) trail.SetAfter(entity); _dbContext.Add(trail); await _dbContext.SaveChangesAsync(); - _BitweenCache.BroadcastRevoke(); + await _BitweenCache.BroadcastRevoke(); await _broadcast.RefreshConsumers(); return null; } diff --git a/SW.Bitween.Api/Resources/GlobalAdapterValuesSets/Create.cs b/SW.Bitween.Api/Resources/GlobalAdapterValuesSets/Create.cs index 551bc716..0755c0ca 100644 --- a/SW.Bitween.Api/Resources/GlobalAdapterValuesSets/Create.cs +++ b/SW.Bitween.Api/Resources/GlobalAdapterValuesSets/Create.cs @@ -11,11 +11,13 @@ public class Create : ICommandHandler { private readonly BitweenDbContext _dbContext; private readonly RequestContext _requestContext; + private readonly IInfolinkCache _cache; - public Create(BitweenDbContext dbContext, RequestContext requestContext) + public Create(BitweenDbContext dbContext, RequestContext requestContext, IInfolinkCache cache) { _dbContext = dbContext; _requestContext = requestContext; + _cache = cache; } public async Task Handle(GlobalAdapterValuesSetCreate request) @@ -35,6 +37,7 @@ public async Task Handle(GlobalAdapterValuesSetCreate request) _dbContext.Add(entity); await _dbContext.SaveChangesAsync(); + await _cache.BroadcastRevoke(); return new { entity.Id diff --git a/SW.Bitween.Api/Resources/GlobalAdapterValuesSets/Delete.cs b/SW.Bitween.Api/Resources/GlobalAdapterValuesSets/Delete.cs index 40a20d97..b7a5cdcb 100644 --- a/SW.Bitween.Api/Resources/GlobalAdapterValuesSets/Delete.cs +++ b/SW.Bitween.Api/Resources/GlobalAdapterValuesSets/Delete.cs @@ -10,11 +10,13 @@ public class Delete : ICommandHandler Handle(string key, DeleteGlobalAdapterValuesSetModel _) @@ -27,6 +29,7 @@ public async Task Handle(string key, DeleteGlobalAdapterValuesSetModel _ _dbContext.Remove(entity); await _dbContext.SaveChangesAsync(); + await _cache.BroadcastRevoke(); return null; } } diff --git a/SW.Bitween.Api/Resources/GlobalAdapterValuesSets/Update.cs b/SW.Bitween.Api/Resources/GlobalAdapterValuesSets/Update.cs index 46377ad0..cdf57a88 100644 --- a/SW.Bitween.Api/Resources/GlobalAdapterValuesSets/Update.cs +++ b/SW.Bitween.Api/Resources/GlobalAdapterValuesSets/Update.cs @@ -10,11 +10,13 @@ public class Update : ICommandHandler Handle(string key, GlobalAdapterValuesSetUpdate request) @@ -29,6 +31,7 @@ public async Task Handle(string key, GlobalAdapterValuesSetUpdate reques entity.Values = request.Values; await _dbContext.SaveChangesAsync(); + await _cache.BroadcastRevoke(); return null; } diff --git a/SW.Bitween.Api/Resources/Notifiers/Create.cs b/SW.Bitween.Api/Resources/Notifiers/Create.cs index bc6bb088..7af11da4 100644 --- a/SW.Bitween.Api/Resources/Notifiers/Create.cs +++ b/SW.Bitween.Api/Resources/Notifiers/Create.cs @@ -10,11 +10,13 @@ public class Create : ICommandHandler { private readonly BitweenDbContext _dbContext; private readonly RequestContext _requestContext; + private readonly IInfolinkCache _cache; - public Create(BitweenDbContext dbContext, RequestContext requestContext) + public Create(BitweenDbContext dbContext, RequestContext requestContext, IInfolinkCache cache) { this._dbContext = dbContext; _requestContext = requestContext; + _cache = cache; } public async Task Handle(NotifierCreate request) @@ -25,6 +27,7 @@ public async Task Handle(NotifierCreate request) _dbContext.Add(notifier); await _dbContext.SaveChangesAsync(); + await _cache.BroadcastRevoke(); return notifier.Id; } diff --git a/SW.Bitween.Api/Resources/Notifiers/Delete.cs b/SW.Bitween.Api/Resources/Notifiers/Delete.cs index 611e0eb9..bed3632a 100644 --- a/SW.Bitween.Api/Resources/Notifiers/Delete.cs +++ b/SW.Bitween.Api/Resources/Notifiers/Delete.cs @@ -9,11 +9,13 @@ public class Delete : IDeleteHandler { private readonly BitweenDbContext _dbContext; private readonly RequestContext _requestContext; + private readonly IInfolinkCache _cache; - public Delete(BitweenDbContext dbContext, RequestContext requestContext) + public Delete(BitweenDbContext dbContext, RequestContext requestContext, IInfolinkCache cache) { _dbContext = dbContext; _requestContext = requestContext; + _cache = cache; } /// @@ -26,6 +28,7 @@ public async Task Handle(int key) await _requestContext.EnsurePermission(_dbContext, Model.Permissions.Notifiers.Delete); await _dbContext.DeleteByKeyAsync(key); + await _cache.BroadcastRevoke(); return null; } } diff --git a/SW.Bitween.Api/Resources/Notifiers/Update.cs b/SW.Bitween.Api/Resources/Notifiers/Update.cs index 302f4c83..25a7aba9 100644 --- a/SW.Bitween.Api/Resources/Notifiers/Update.cs +++ b/SW.Bitween.Api/Resources/Notifiers/Update.cs @@ -11,11 +11,13 @@ public class Update : ICommandHandler { private readonly BitweenDbContext _dbContext; private readonly RequestContext _requestContext; + private readonly IInfolinkCache _cache; - public Update(BitweenDbContext dbContext, RequestContext requestContext) + public Update(BitweenDbContext dbContext, RequestContext requestContext, IInfolinkCache cache) { _dbContext = dbContext; _requestContext = requestContext; + _cache = cache; } public async Task Handle(int key, NotifierUpdate request) @@ -37,6 +39,7 @@ public async Task Handle(int key, NotifierUpdate request) await _dbContext.SaveChangesAsync(); + await _cache.BroadcastRevoke(); return null; } diff --git a/SW.Bitween.Api/Resources/Subscriptions/Delete.cs b/SW.Bitween.Api/Resources/Subscriptions/Delete.cs index 2fe856e4..4ae1a368 100644 --- a/SW.Bitween.Api/Resources/Subscriptions/Delete.cs +++ b/SW.Bitween.Api/Resources/Subscriptions/Delete.cs @@ -13,12 +13,14 @@ public class Delete : IDeleteHandler { private readonly BitweenDbContext _dbContext; private readonly RequestContext _requestContext; + private readonly IInfolinkCache _cache; - public Delete(BitweenDbContext dbContext, RequestContext requestContext) + public Delete(BitweenDbContext dbContext, RequestContext requestContext, IInfolinkCache cache) { this._dbContext = dbContext; _requestContext = requestContext; + _cache = cache; } public async Task Handle(int key) @@ -28,6 +30,7 @@ public async Task Handle(int key) await EnsureNothingPointsAtIt(key); await _dbContext.DeleteByKeyAsync(key); + await _cache.BroadcastRevoke(); return null; } diff --git a/SW.Bitween.Api/Resources/Subscriptions/Pause.cs b/SW.Bitween.Api/Resources/Subscriptions/Pause.cs index c56317ff..b86af1c2 100644 --- a/SW.Bitween.Api/Resources/Subscriptions/Pause.cs +++ b/SW.Bitween.Api/Resources/Subscriptions/Pause.cs @@ -12,12 +12,14 @@ public class Pause : ICommandHandler { private readonly BitweenDbContext _dbContext; private readonly RequestContext _requestContext; + private readonly IInfolinkCache _cache; - public Pause(BitweenDbContext dbContext, RequestContext requestContext) + public Pause(BitweenDbContext dbContext, RequestContext requestContext, IInfolinkCache cache) { _dbContext = dbContext; _requestContext = requestContext; + _cache = cache; } public async Task Handle(int key, SubscriptionPause request) @@ -40,6 +42,11 @@ public async Task Handle(int key, SubscriptionPause request) trail.SetAfter(entity); _dbContext.Add(trail); await _dbContext.SaveChangesAsync(); + // The receiving path reads PausedOn off the cached copy, so without this a paused + // integration keeps taking messages for the rest of the cache's ten minutes. Resuming + // has the mirror problem: its handler re-reads the cache, finds the copy still paused + // and returns early, leaving everything it held on hold. + await _cache.BroadcastRevoke(); return new { entity.Id diff --git a/SW.Bitween.Api/Resources/Subscriptions/SaveMapper.cs b/SW.Bitween.Api/Resources/Subscriptions/SaveMapper.cs index 7819dd5a..4143203c 100644 --- a/SW.Bitween.Api/Resources/Subscriptions/SaveMapper.cs +++ b/SW.Bitween.Api/Resources/Subscriptions/SaveMapper.cs @@ -39,7 +39,7 @@ public async Task Handle(int key, SubscriptionSaveMapper model) ); await _dbContext.SaveChangesAsync(); - _BitweenCache.BroadcastRevoke(); + await _BitweenCache.BroadcastRevoke(); return null; } diff --git a/SW.Bitween.Api/Resources/WorkGroups/Create.cs b/SW.Bitween.Api/Resources/WorkGroups/Create.cs index 931f2699..e8c813e6 100644 --- a/SW.Bitween.Api/Resources/WorkGroups/Create.cs +++ b/SW.Bitween.Api/Resources/WorkGroups/Create.cs @@ -30,7 +30,7 @@ public async Task Handle(CreateWorkGroupModel request) }; dbContext.Add(workgroup); await dbContext.SaveChangesAsync(); - _BitweenCache.BroadcastRevoke(); + await _BitweenCache.BroadcastRevoke(); await _broadcast.RefreshConsumers(); return new { diff --git a/SW.Bitween.Api/Resources/WorkGroups/Delete.cs b/SW.Bitween.Api/Resources/WorkGroups/Delete.cs index 23db025a..17a65f13 100644 --- a/SW.Bitween.Api/Resources/WorkGroups/Delete.cs +++ b/SW.Bitween.Api/Resources/WorkGroups/Delete.cs @@ -26,7 +26,7 @@ public async Task Handle(int key, DeleteWorkGroupModel _) //Todo chek rabbitMq dbContext.Remove(category); await dbContext.SaveChangesAsync(); - _infolinkCache.BroadcastRevoke(); + await _infolinkCache.BroadcastRevoke(); await _broadcast.RefreshConsumers(); return null; } diff --git a/SW.Bitween.Api/Services/Caching/InMemoryInfolinkCache.cs b/SW.Bitween.Api/Services/Caching/InMemoryInfolinkCache.cs index df5be0f7..829880ef 100644 --- a/SW.Bitween.Api/Services/Caching/InMemoryInfolinkCache.cs +++ b/SW.Bitween.Api/Services/Caching/InMemoryInfolinkCache.cs @@ -205,5 +205,8 @@ public void Revoke() _cache.Remove(nameof(Document)); _cache.Remove(nameof(WorkGroup)); _cache.Remove(nameof(BusGateway)); + // Load() caches this one too. Leaving it out here meant no write of any kind could + // clear a global value: it sat for its full ten minutes regardless. + _cache.Remove(nameof(GlobalAdapterValuesSet)); } } \ No newline at end of file diff --git a/SW.Bitween.IntegrationTests/Tests/CacheRevocationTests.cs b/SW.Bitween.IntegrationTests/Tests/CacheRevocationTests.cs new file mode 100644 index 00000000..31663f21 --- /dev/null +++ b/SW.Bitween.IntegrationTests/Tests/CacheRevocationTests.cs @@ -0,0 +1,134 @@ +using System; +using System.Threading; +using System.Threading.Tasks; +using Microsoft.Extensions.DependencyInjection; +using SW.Bitween.Domain; +using SW.Bitween.Domain.Gateway; +using SW.Bitween.IntegrationTests.Fixtures; +using SW.Bitween.Model; +using Xunit; + +namespace SW.Bitween.IntegrationTests.Tests; + +/// +/// The server holds subscriptions, information types, notifiers, work groups, global values and +/// bus gateways in memory for ten minutes, because the message path reads them for every message +/// and cannot go to the database each time. A write therefore has to announce itself, or the +/// running system keeps acting on what it read ten minutes ago. +/// +/// +/// Announcing means BroadcastRevoke(), which publishes rather than clearing directly: each +/// instance owns a private queue bound to the shared node exchange, so every instance clears its +/// own copy — including the one that handled the write. The fixture deliberately does not call +/// AddBusConsume, so that round trip cannot complete here and the announcement is asserted +/// at the handler instead, with a cache that records the call. +/// +[Collection("Bitween")] +public class CacheRevocationTests +{ + private readonly BitweenFixture _fixture; + + public CacheRevocationTests(BitweenFixture fixture) + { + _fixture = fixture; + } + + private static int _seq; + private static string Unique(string prefix) => $"{prefix}-{Interlocked.Increment(ref _seq)}"; + + [Fact] + public async Task Pausing_announces_the_write_so_the_receiving_path_stops_seeing_it_as_running() + { + int subscriptionId; + await using (var scope = _fixture.CreateScope()) + { + var db = scope.ServiceProvider.GetRequiredService(); + var document = new Document(null, Unique("Pause revoke doc"), DocumentFormat.Json); + db.Set().Add(document); + await db.SaveChangesAsync(); + + var subscription = new Subscription(Unique("Pause revoke"), document.Id) { Inactive = false }; + db.Set().Add(subscription); + await db.SaveChangesAsync(); + subscriptionId = subscription.Id; + } + + var recorder = new RecordingCache(); + await using (var scope = _fixture.CreateScope()) + { + scope.Superuser(); + var pause = ActivatorUtilities.CreateInstance( + scope.ServiceProvider, recorder); + await pause.Handle(subscriptionId, new SubscriptionPause()); + } + + // Without this, XchangeService keeps reading PausedOn off a copy taken before the pause and + // goes on creating ordinary exchanges for an integration the operator has stopped. Resuming + // has the mirror failure: the handler finds its own cached copy still paused and returns + // without releasing anything it held. + Assert.Equal(1, recorder.Broadcasts); + } + + [Fact] + public async Task Revoking_clears_global_values_too() + { + var cache = _fixture.App.Services.GetRequiredService(); + var id = Unique("global-revoke"); + + await using (var scope = _fixture.CreateScope()) + { + var db = scope.ServiceProvider.GetRequiredService(); + db.Set().Add(new GlobalAdapterValuesSet + { + Id = id, + Name = "Before", + Values = new(), + }); + await db.SaveChangesAsync(); + } + + cache.Revoke(); + Assert.Equal("Before", (await cache.GlobalAdapterValuesSetById(id))?.Name); + + await using (var scope = _fixture.CreateScope()) + { + var db = scope.ServiceProvider.GetRequiredService(); + var entity = await db.Set().FindAsync(id); + entity!.Name = "After"; + await db.SaveChangesAsync(); + } + + // Load() caches global values but Revoke() used to skip them, so this read returned + // "Before" no matter what had been revoked, until the ten minutes were up. + cache.Revoke(); + Assert.Equal("After", (await cache.GlobalAdapterValuesSetById(id))?.Name); + } + + /// + /// Counts announcements and refuses everything else, so a handler that starts reading through + /// the cache fails here rather than quietly passing against a stub that answered. + /// + private sealed class RecordingCache : IInfolinkCache + { + public int Broadcasts { get; private set; } + + public Task BroadcastRevoke() + { + Broadcasts++; + return Task.CompletedTask; + } + + public void Revoke() => throw new NotSupportedException(); + public Task ListSubscriptionsByDocumentAsync(int documentId) => throw new NotSupportedException(); + public Task ListBusGatewayRoutesByDocumentAsync(int documentId) => throw new NotSupportedException(); + public Task ListNotifiersAsync() => throw new NotSupportedException(); + public Task SubscriptionByIdAsync(int subscriptionId) => throw new NotSupportedException(); + public Task DocumentByIdAsync(int documentId) => throw new NotSupportedException(); + public Task DocumentByNameAsync(string documentName) => throw new NotSupportedException(); + public Task ListWorkGroupsAsync() => throw new NotSupportedException(); + public Task WorkGroupByIdAsync(int workGroupId) => throw new NotSupportedException(); + public Task WorkGroupBySubscriptionIdAsync(int subscriptionId) => throw new NotSupportedException(); + public Task GlobalAdapterValuesSetById(string id) => throw new NotSupportedException(); + public Task ListGlobalAdapterValuesSetsAsync() => throw new NotSupportedException(); + } +} From 698e8874bffdd65fa48958183d7cdabba350364c Mon Sep 17 00:00:00 2001 From: Hamza Alqurneh Date: Sun, 30 Aug 2026 13:04:57 +0300 Subject: [PATCH 2/2] fix: don't let an in-flight load republish revoked data MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Load() runs six database reads and then six writes, and Revoke() runs on the bus consumer thread against the same singleton. A revoke landing between the two is revoking the snapshot those reads just took, so publishing it anyway put the staleness straight back for the full ten minutes — the failure this branch exists to prevent. Revoke() now bumps a generation before clearing, and a load whose generation moved reads again instead of publishing. Callers rely on the cache being populated when Load() returns, so the last of three attempts publishes regardless. Raised by CodeRabbit on #272 against the global-values path; it applies to all six cached sets. --- .../Services/Caching/InMemoryInfolinkCache.cs | 77 ++++++++++++++----- 1 file changed, 59 insertions(+), 18 deletions(-) diff --git a/SW.Bitween.Api/Services/Caching/InMemoryInfolinkCache.cs b/SW.Bitween.Api/Services/Caching/InMemoryInfolinkCache.cs index 829880ef..b95308fa 100644 --- a/SW.Bitween.Api/Services/Caching/InMemoryInfolinkCache.cs +++ b/SW.Bitween.Api/Services/Caching/InMemoryInfolinkCache.cs @@ -1,5 +1,6 @@ using System; using System.Linq; +using System.Threading; using System.Threading.Tasks; using Microsoft.EntityFrameworkCore; using Microsoft.Extensions.Caching.Memory; @@ -21,6 +22,12 @@ public class InMemoryBitweenCache : IInfolinkCache private readonly IServiceScopeFactory _ssf; private readonly ILogger _logger; + /// How many times re-reads before publishing regardless. + private const int MaxLoadAttempts = 3; + + /// Bumped by every , so a load can tell one overtook it. + private long _generation; + public InMemoryBitweenCache(IMemoryCache memoryCache, IServiceScopeFactory ssf, ILogger logger) @@ -30,26 +37,55 @@ public InMemoryBitweenCache(IMemoryCache memoryCache, IServiceScopeFactory ssf, _logger = logger ?? throw new ArgumentNullException(nameof(logger)); } + /// + /// Reads every cached set from the database and publishes it. + /// + /// + /// The six reads take time, and runs on the bus consumer thread against + /// this same singleton. A revoke landing between the reads and the writes is revoking the + /// snapshot those reads just took — the write it announces happened after they began — so + /// publishing it anyway would reinstate the staleness the revoke existed to clear, for the + /// full ten minutes. Hence the generation check: read again rather than publish a snapshot + /// that is already known to be behind. + /// + /// Callers rely on this having populated the cache by the time it returns, so the last attempt + /// publishes regardless. Two revokes landing inside one set of reads is already unlikely; three + /// means the system is revoking continuously, and making progress matters more than the last + /// few milliseconds of freshness. + /// private async Task Load() { - using var scope = _ssf.CreateScope(); - var repo = scope.ServiceProvider.GetRequiredService(); - _logger.LogInformation("Loading documents and subscriptions to cache"); - var cachedSubscriptions = await repo.Set().Include(s=>s.WorkGroup) - .AsNoTracking().Where(i => !i.Inactive).ToArrayAsync(); - var cachedDocuments = await repo.Set().AsNoTracking().ToArrayAsync(); - var cachedNotifiers = await repo.Set().Where(i => !i.Inactive).AsNoTracking().ToArrayAsync(); - var cachedWorkGroups = await repo.Set().AsNoTracking().ToArrayAsync(); - var cachedGlobalValues = await repo.Set().AsNoTracking().ToArrayAsync(); - var cachedBusGateways = await repo.Set().Include(g => g.Routes).AsNoTracking().ToArrayAsync(); - var span = TimeSpan.FromMinutes(10); - _cache.Set(nameof(Document), cachedDocuments, span); - - _cache.Set(nameof(Subscription), cachedSubscriptions, span); - _cache.Set(nameof(Notifier), cachedNotifiers, span); - _cache.Set(nameof(WorkGroup), cachedWorkGroups, span); - _cache.Set(nameof(GlobalAdapterValuesSet), cachedGlobalValues, span); - _cache.Set(nameof(BusGateway), cachedBusGateways, span); + for (var attempt = 0; ; attempt++) + { + var generation = Volatile.Read(ref _generation); + + using var scope = _ssf.CreateScope(); + var repo = scope.ServiceProvider.GetRequiredService(); + _logger.LogInformation("Loading documents and subscriptions to cache"); + var cachedSubscriptions = await repo.Set().Include(s=>s.WorkGroup) + .AsNoTracking().Where(i => !i.Inactive).ToArrayAsync(); + var cachedDocuments = await repo.Set().AsNoTracking().ToArrayAsync(); + var cachedNotifiers = await repo.Set().Where(i => !i.Inactive).AsNoTracking().ToArrayAsync(); + var cachedWorkGroups = await repo.Set().AsNoTracking().ToArrayAsync(); + var cachedGlobalValues = await repo.Set().AsNoTracking().ToArrayAsync(); + var cachedBusGateways = await repo.Set().Include(g => g.Routes).AsNoTracking().ToArrayAsync(); + + if (Volatile.Read(ref _generation) != generation && attempt < MaxLoadAttempts - 1) + { + _logger.LogInformation("Cache was revoked while loading; reading again"); + continue; + } + + var span = TimeSpan.FromMinutes(10); + _cache.Set(nameof(Document), cachedDocuments, span); + + _cache.Set(nameof(Subscription), cachedSubscriptions, span); + _cache.Set(nameof(Notifier), cachedNotifiers, span); + _cache.Set(nameof(WorkGroup), cachedWorkGroups, span); + _cache.Set(nameof(GlobalAdapterValuesSet), cachedGlobalValues, span); + _cache.Set(nameof(BusGateway), cachedBusGateways, span); + return; + } } public async Task ListSubscriptionsByDocumentAsync(int documentId) @@ -200,6 +236,11 @@ public async Task ListGlobalAdapterValuesSetsAsync() public void Revoke() { + // Before the removals, so a load that reads after this sees the new value and a load that + // read before it is discarded. Bumping afterwards would leave a window where a load could + // both miss the removal and match the generation. + Interlocked.Increment(ref _generation); + _cache.Remove(nameof(Subscription)); _cache.Remove(nameof(Notifier)); _cache.Remove(nameof(Document));