diff --git a/SW.Bitween.Api/Resources/Adapters/AdapterListing.cs b/SW.Bitween.Api/Resources/Adapters/AdapterListing.cs index 47294c37..d9f5e762 100644 --- a/SW.Bitween.Api/Resources/Adapters/AdapterListing.cs +++ b/SW.Bitween.Api/Resources/Adapters/AdapterListing.cs @@ -40,6 +40,13 @@ public class AdapterListing( Services.Adapters.AdapterCatalog catalog, Services.Adapters.NativeAdapterManifests nativeManifests) { + /// + /// How many adapters' metadata the lists may be reading from storage at once, across the whole + /// process: a screen asks for its four kinds together, each reads every other kind's adapters, + /// and storage services limit their callers. + /// + private static readonly System.Threading.SemaphoreSlim MetadataReads = new(16, 16); + /// The plural, lowercase kind: receivers, handlers, … /// Native adapters first, then the published ones. public async Task> List(string prefix) @@ -147,35 +154,49 @@ private async Task> ListByKindAsync(string prefix) var kind = prefix?.TrimEnd('s') ?? ""; if (string.IsNullOrWhiteSpace(kind)) return byConvention; - foreach (var item in await cloudFilesService.ListAsync($"{root}/")) - { - if (item.Size <= 0 || named.Contains(item.Key)) continue; - - var declared = await DeclaredKindsAsync(item.Key, root); - if (declared.Contains(kind, System.StringComparer.OrdinalIgnoreCase)) - byConvention.Add(item); - } + var others = (await cloudFilesService.ListAsync($"{root}/")) + .Where(item => item.Size > 0 && !named.Contains(item.Key)) + .ToList(); + // Side by side: once the installer's metadata cache runs out, five minutes by default, this + // reads every other kind's adapters from storage, and one after another that took half a + // minute for a few dozen. Each adapter once, however many of its versions are listed. + var declared = (await Task.WhenAll(others.Select(i => AdapterIdOf(i.Key, root)).Distinct(System.StringComparer.Ordinal) + .Select(async adapterId => + { + await MetadataReads.WaitAsync(); + try { return (Id: adapterId, Kinds: await DeclaredKindsAsync(adapterId)); } + finally { MetadataReads.Release(); } + }))) + .ToDictionary(d => d.Id, d => d.Kinds, System.StringComparer.Ordinal); + + byConvention.AddRange(others.Where(item => + declared[AdapterIdOf(item.Key, root)].Contains(kind, System.StringComparer.OrdinalIgnoreCase))); return byConvention; } + /// The adapter a listed file is a package of. + private static string AdapterIdOf(string key, string root) + { + var adapterId = key.StartsWith($"{root}/", System.StringComparison.OrdinalIgnoreCase) + ? key[(root.Length + 1)..] + : key; + + // Versioned uploads keep the adapter id one segment up from the version. + return Semver.IsVersionNumber(adapterId.Split('/').Last()) + ? string.Join('/', adapterId.Split('/')[..^1]) + : adapterId; + } + /// /// The kinds one adapter declared. Metadata reads are cached by the installer, and an adapter /// whose metadata cannot be read declares nothing rather than taking the whole catalogue down /// with it - the list is what an operator needs to configure anything at all. /// - private async Task DeclaredKindsAsync(string key, string root) + private async Task DeclaredKindsAsync(string adapterId) { try { - var adapterId = key.StartsWith($"{root}/", System.StringComparison.OrdinalIgnoreCase) - ? key[(root.Length + 1)..] - : key; - - // Versioned uploads keep the adapter id one segment up from the version. - if (Semver.IsVersionNumber(adapterId.Split('/').Last())) - adapterId = string.Join('/', adapterId.Split('/')[..^1]); - var metadata = await adapterInstaller.GetMetadataAsync(adapterId); if (metadata?.AdapterValues == null) return []; diff --git a/SW.Bitween.Api/Services/ServerlessAdapterDescriber.cs b/SW.Bitween.Api/Services/ServerlessAdapterDescriber.cs index d5cd47b4..86bf855c 100644 --- a/SW.Bitween.Api/Services/ServerlessAdapterDescriber.cs +++ b/SW.Bitween.Api/Services/ServerlessAdapterDescriber.cs @@ -17,8 +17,9 @@ namespace SW.Bitween; /// /// Asking is expensive out of all proportion to the answer: the package is downloaded and unzipped, /// a dotnet child process is started, and the question goes over stdio - to be told a handful -/// of key names. So the answer is cached, one ask is shared by everyone who wants it at that moment, -/// and only so many adapters may be running at once. +/// of key names. So the answer is remembered for as long as the build it came from is the one that +/// runs, one ask is shared by everyone who wants it at that moment, and only so many adapters may be +/// running at once. /// /// /// A singleton, and it has to be. All three of those only work between requests: a per-request copy @@ -28,7 +29,7 @@ namespace SW.Bitween; public class ServerlessAdapterDescriber( IServiceScopeFactory scopeFactory, IMemoryCache memoryCache, - ServerlessOptions serverlessOptions) + SW.Serverless.AdapterInstaller adapterInstaller) { private const string CacheKeyPrefix = "bitween.adapters.startupvalues"; @@ -53,14 +54,16 @@ public class ServerlessAdapterDescriber( public async Task> Describe(string adapterId) { - if (memoryCache.TryGetValue(CacheKey(adapterId), out IDictionary cached)) - return cached; + var build = await BuildOf(adapterId); + if (memoryCache.TryGetValue(CacheKey(adapterId), out Answer cached) && cached.Build == build) + return cached.Values; // Everyone who wants this adapter while it is being asked waits on the one ask, rather than // starting an identical child process of their own. Without this the cache does not help // the case it most needs to - a cold start, when every request arrives at once. - var asking = _asking.GetOrAdd(adapterId, id => - new Lazy>>(() => Ask(id), + var key = $"{adapterId}@{build}"; + var asking = _asking.GetOrAdd(key, _ => + new Lazy>>(() => Ask(adapterId, build), LazyThreadSafetyMode.ExecutionAndPublication)); try @@ -74,18 +77,34 @@ public async Task> Describe(string adapterId) // needed this entry anyway. Matched on the instance so that a newer ask, started by // someone else after this one finished, is not the one taken away. _asking.TryRemove(new KeyValuePair>>>( - adapterId, asking)); + key, asking)); } } - private async Task> Ask(string adapterId) + /// + /// Which build of the adapter a start would run: its package's hash, as the installer has it. + /// A storage read at most, where asking is a child process. + /// + private async Task BuildOf(string adapterId) + { + try + { + return (await adapterInstaller.GetMetadataAsync(adapterId)).Hash; + } + catch (KeyNotFoundException ex) + { + throw Incomplete(adapterId, ex); + } + } + + private async Task> Ask(string adapterId, string build) { await _starts.WaitAsync(); try { // The queue may have been long enough for someone else to have answered this already. - if (memoryCache.TryGetValue(CacheKey(adapterId), out IDictionary cached)) - return cached; + if (memoryCache.TryGetValue(CacheKey(adapterId), out Answer cached) && cached.Build == build) + return cached.Values; // Its own scope, disposed the moment this returns, because disposing the serverless // service is what quits the child process. On the request's scope instead, a caller @@ -100,9 +119,7 @@ private async Task> Ask(string adapterId) } catch (KeyNotFoundException ex) { - throw new BitweenException( - $"Adapter '{adapterId}' metadata is incomplete or the adapter package is not installed. " + - $"Missing metadata key: {ex.Message}", ex); + throw Incomplete(adapterId, ex); } var startupValues = await serverless.GetExpectedStartupValues() @@ -117,11 +134,14 @@ private async Task> Ask(string adapterId) // Only a successful answer is cached - a boot that failed because, say, the adapter's // runtime is missing locally must not stick around as "this adapter has no properties". // - // Held for as long as ServerlessService already serves a stale Hash from its own - // metadata cache. Within that window it boots the previous build of a re-uploaded - // adapter regardless, so this adds no staleness that re-uploading did not already have. - return memoryCache.Set(CacheKey(adapterId), shared, - TimeSpan.FromMinutes(serverlessOptions.AdapterMetadataCacheDuration)); + // Held for as long as this build is the one that runs, rather than the five minutes it + // was, after which every adapter was started again for an answer that had not changed: + // the first read of a subscription after that took a second per adapter. A re-uploaded + // adapter has a new hash, which the installer has once its metadata cache runs out - + // when ServerlessService starts booting the new build too - and is asked again then. A + // day without anyone asking lets the answer go. + return memoryCache.Set(CacheKey(adapterId), new Answer(build, shared), + new MemoryCacheEntryOptions { SlidingExpiration = TimeSpan.FromDays(1) }).Values; } finally { @@ -129,6 +149,13 @@ private async Task> Ask(string adapterId) } } + private static BitweenException Incomplete(string adapterId, KeyNotFoundException ex) => new( + $"Adapter '{adapterId}' metadata is incomplete or the adapter package is not installed. " + + $"Missing metadata key: {ex.Message}", ex); + + /// What one build of an adapter said it expects. + private sealed record Answer(string Build, IDictionary Values); + private static string CacheKey(string adapterId) => $"{CacheKeyPrefix}.{adapterId.ToLowerInvariant()}"; } diff --git a/SW.Bitween.IntegrationTests/Tests/AdapterCatalogTests.cs b/SW.Bitween.IntegrationTests/Tests/AdapterCatalogTests.cs index f94a4d77..e0ee0e5d 100644 --- a/SW.Bitween.IntegrationTests/Tests/AdapterCatalogTests.cs +++ b/SW.Bitween.IntegrationTests/Tests/AdapterCatalogTests.cs @@ -1,9 +1,11 @@ using System; using System.Collections.Generic; +using System.IO; using System.Linq; using System.Threading; using System.Threading.Tasks; using Microsoft.EntityFrameworkCore; +using Microsoft.Extensions.Caching.Memory; using Microsoft.Extensions.DependencyInjection; using Newtonsoft.Json; using Newtonsoft.Json.Linq; @@ -183,6 +185,55 @@ public async Task An_adapter_published_the_old_way_lists_describes_and_runs_as_b Assert.True(result.Success, result.Exception); } + [Fact] + public async Task An_adapter_is_started_to_describe_it_once_a_build() + { + // No manifest, so describing it means starting it. + var id = await PublishHandler(); + + await using var scope = fixture.CreateScope(); + var services = scope.ServiceProvider; + var cloudFiles = services.GetRequiredService(); + var clock = new TestClock(); + var starts = new CountingScopes(services.GetRequiredService()); + var installer = new SW.Serverless.AdapterInstaller(services.GetRequiredService(), + new MemoryCache(new MemoryCacheOptions { Clock = clock }), cloudFiles); + var describer = new ServerlessAdapterDescriber(starts, new MemoryCache(new MemoryCacheOptions { Clock = clock }), installer); + + await describer.Describe(id); + // Long past the five minutes an answer used to last, and the installer's metadata cache with it. + clock.Advance(TimeSpan.FromMinutes(30)); + Assert.True((await describer.Describe(id)).ContainsKey("ContentType")); + Assert.Equal(1, starts.Created); + + // A new build is asked about once the installer has it. + await AdapterInstaller.InstallAsync(cloudFiles, SampleProject, id, SampleEntry, + extraFiles: new Dictionary { ["rebuilt.txt"] = "a new build" }); + clock.Advance(TimeSpan.FromMinutes(6)); + await describer.Describe(id); + Assert.Equal(2, starts.Created); + } + + /// A clock the test moves on itself. + sealed class TestClock : Microsoft.Extensions.Internal.ISystemClock + { + public DateTimeOffset UtcNow { get; private set; } = DateTimeOffset.UtcNow; + public void Advance(TimeSpan by) => UtcNow += by; + } + + /// Counts the scopes it makes: the describer makes one for each adapter it starts. + sealed class CountingScopes(IServiceScopeFactory inner) : IServiceScopeFactory + { + int _created; + public int Created => _created; + + public IServiceScope CreateScope() + { + Interlocked.Increment(ref _created); + return inner.CreateScope(); + } + } + [Fact] public async Task Catalog_entries_are_never_listed_as_adapters() { @@ -246,6 +297,69 @@ public async Task Asked_without_startup_values_the_catalog_lists_versions_and_st Assert.Equal(JTokenType.Null, listed["startupValues"]!.Type); } + [Fact] + public async Task Listing_a_kind_reads_the_other_adapters_side_by_side() + { + // Named for no kind, so a list of handlers only learns what they are from each one's metadata. + var ids = Enumerable.Range(0, 6).Select(_ => $"acme.sidebyside{Interlocked.Increment(ref _seq)}").ToList(); + await using (var scope = fixture.CreateScope()) + { + var cloudFiles = scope.ServiceProvider.GetRequiredService(); + foreach (var id in ids) + await AdapterInstaller.InstallAsync(cloudFiles, SampleProject, id, SampleEntry, + new Dictionary { ["Kind"] = "handler" }); + } + + await using var listScope = fixture.CreateScope(); + var services = listScope.ServiceProvider; + var storage = new SlowStorage(services.GetRequiredService()); + // An empty metadata cache of its own, as the installer's is every five minutes. + var installer = new SW.Serverless.AdapterInstaller(services.GetRequiredService(), + new MemoryCache(new MemoryCacheOptions()), storage); + var listing = ActivatorUtilities.CreateInstance(services, storage, installer); + + var keys = (await listing.List("handlers")).Select(a => a.Key).ToList(); + + Assert.All(ids, id => Assert.Contains(id, keys)); + Assert.True(storage.MostAtOnce > 1, "Each metadata read waited for the one before it."); + } + + /// + /// Storage that answers lists and metadata reads after a delay, as a remote one does, and keeps + /// the most it was answering at once. + /// + sealed class SlowStorage(ICloudFilesService inner) : ICloudFilesService + { + readonly object _gate = new(); + int _answering; + public int MostAtOnce { get; private set; } + + async Task Slowly(Func> read) + { + lock (_gate) MostAtOnce = Math.Max(MostAtOnce, ++_answering); + try + { + await Task.Delay(50); + return await read(); + } + finally + { + lock (_gate) _answering--; + } + } + + public Task> ListAsync(string prefix) => Slowly(() => inner.ListAsync(prefix)); + public Task> GetMetadataAsync(string key) => Slowly(() => inner.GetMetadataAsync(key)); + + public Task WriteAsync(Stream inputStream, WriteFileSettings settings) => inner.WriteAsync(inputStream, settings); + public Task WriteTextAsync(string text, WriteFileSettings settings) => inner.WriteTextAsync(text, settings); + public string GetSignedUrl(string key, TimeSpan expiry) => inner.GetSignedUrl(key, expiry); + public string GetUrl(string key) => inner.GetUrl(key); + public WriteWrapper OpenWrite(WriteFileSettings settings) => inner.OpenWrite(settings); + public Task OpenReadAsync(string key) => inner.OpenReadAsync(key); + public Task DeleteAsync(string key) => inner.DeleteAsync(key); + } + [Fact] public async Task Native_adapters_list_with_their_manifest() { diff --git a/docs/caveats.md b/docs/caveats.md index 1803218f..3f65ca28 100644 --- a/docs/caveats.md +++ b/docs/caveats.md @@ -29,7 +29,7 @@ These behaviours were found while documenting Bitween from its source code, and | The dashboard's *Failures to act on* tile says it covers 14 days, but counts failures of any age. | `ClientApp/src/pages/dashboard/DashboardPage.tsx` | | An exchange's delivered time is never recorded. | `Domain/XchangeDelivery.cs` | | Exchanges, results and notifications are kept for ever by default (`ExchangeRetentionDays` 0), while their files expire from storage. The retention panel on the Settings page warns when exchanges outlive their files. | `Services/Retention/ExchangeRetentionJob.cs` | -| Cached configuration can stay stale on other instances for up to 10 minutes if the revoke broadcast fails. Adapter descriptions and catalog entries are cached per node; publishing, promoting or withdrawing through Bitween clears them everywhere over the bus, but a package published straight to storage is seen with its old properties for up to a minute. | `Services/Caching/InMemoryInfolinkCache.cs`, `Services/ServerlessAdapterDescriber.cs` | +| Cached configuration can stay stale on other instances for up to 10 minutes if the revoke broadcast fails. Adapter descriptions and catalog entries are cached per node; publishing, promoting or withdrawing through Bitween clears them everywhere over the bus, but a package published straight to storage is seen with its old properties for up to a minute, or five for one without a manifest, which is described by starting it. That answer is kept until the adapter's package changes, so it is started once a build rather than every five minutes. | `Services/Caching/InMemoryInfolinkCache.cs`, `Services/ServerlessAdapterDescriber.cs` | ## Adapters