diff --git a/src/Oddin.OddsFeedSdk/Oddin.OddsFeedSdk.Tests/API/CacheObserverResilienceTests.cs b/src/Oddin.OddsFeedSdk/Oddin.OddsFeedSdk.Tests/API/CacheObserverResilienceTests.cs new file mode 100644 index 0000000..c6c2023 --- /dev/null +++ b/src/Oddin.OddsFeedSdk/Oddin.OddsFeedSdk.Tests/API/CacheObserverResilienceTests.cs @@ -0,0 +1,280 @@ +using System; +using System.Collections.Generic; +using System.Globalization; +using System.Net; +using System.Reactive.Linq; +using System.Reactive.Subjects; +using System.Reflection; +using Oddin.OddsFeedSdk.API; +using Oddin.OddsFeedSdk.API.Abstractions; +using Oddin.OddsFeedSdk.API.Entities; +using Oddin.OddsFeedSdk.API.Models; +using Oddin.OddsFeedSdk.Common; +using Xunit; + +namespace Oddin.OddsFeedSdk.Tests.API; + +// Rx disposes the subscription permanently when an observer throws, and Subject +// walks observers in order without catching, so later caches miss the response too. +[Collection(CacheSideLoadCollection.Name)] +public class CacheObserverResilienceTests +{ + private static readonly CultureInfo Culture = new("en"); + + private static URN PlayerId => new("od:player:1"); + private static URN CompetitorId => new("od:competitor:1"); + private static URN MatchId => new("od:match:1"); + private static URN SportId => new("od:sport:1"); + private static URN TournamentId => new("od:tournament:1"); + + // A payload that throws inside each cache's observer, then a good one. The + // subscription has to survive the first for the second to land. + [Fact] + public void PlayerCacheKeepsSideLoadingAfterAResponseThatThrows() + { + var (api, proxy) = NewApi(); + using var cache = new PlayerCache(api); + + var escaped = Publish(proxy, new competitorProfileEndpoint { players = null }); + Publish(proxy, Profile(new player_profilePlayer { id = "od:player:1", name = "Dendi" })); + + Assert.False(escaped, EscapeMessage); + Assert.NotNull(cache.GetPlayer(PlayerId, new[] { Culture })); + } + + [Fact] + public void CompetitorCacheKeepsSideLoadingAfterAResponseThatThrows() + { + var (api, proxy) = NewApi(); + using var cache = new CompetitorCache(api); + + var escaped = Publish(proxy, new MatchSummaryModel { sport_event = new sportEvent { id = "od:match:1" } }); + Publish(proxy, new TournamentInfoModel + { + tournament = new tournamentExtended { id = "od:tournament:1", competitors = Array.Empty() }, + competitors = new[] { new team { id = "od:competitor:1", name = "Team Secret" } } + }); + + Assert.False(escaped, EscapeMessage); + Assert.NotNull(cache.GetCompetitor(CompetitorId, new[] { Culture })); + } + + [Fact] + public void MatchStatusCacheKeepsSideLoadingAfterAResponseThatThrows() + { + var (api, proxy) = NewApi(); + using var cache = new MatchStatusCache(api); + + // sport_event_status is null, and RefreshOrInsertApiItem dereferences status. + var escaped = Publish(proxy, new MatchSummaryModel { sport_event = new sportEvent { id = "od:match:1" } }); + Publish(proxy, new MatchSummaryModel + { + sport_event = new sportEvent { id = "od:match:1" }, + sport_event_status = new sportEventStatus { status = "live", match_status_code = 1 } + }); + + Assert.False(escaped, EscapeMessage); + Assert.NotNull(cache.GetMatchStatus(MatchId)); + } + + [Fact] + public void SportDataCacheKeepsSideLoadingAfterAResponseThatThrows() + { + var (api, proxy) = NewApi(); + using var cache = new SportDataCache(api); + + // tournament is null, so the switch arm dereferences it building the dictionary. + var escaped = Publish(proxy, new TournamentInfoModel { tournament = null }); + Publish(proxy, TournamentInfo()); + + Assert.False(escaped, EscapeMessage); + Assert.NotNull(ReadSport(cache)); + } + + [Fact] + public void TournamentsCacheKeepsSideLoadingAfterAResponseThatThrows() + { + var (api, proxy) = NewApi(); + using var cache = new TournamentsCache(api); + + var escaped = Publish(proxy, new FixturesEndpointModel { fixture = null }); + Publish(proxy, new TournamentsModel + { + tournaments = new List { new() { id = "od:tournament:1", name = "The International" } } + }); + + Assert.False(escaped, EscapeMessage); + Assert.NotNull(cache.GetTournament(TournamentId, new[] { Culture })); + } + + [Fact] + public void MatchCacheKeepsSideLoadingAfterAResponseThatThrows() + { + var (api, proxy) = NewApi(); + using var cache = new MatchCache(api); + + var escaped = Publish(proxy, new ScheduleEndpointModel { sport_event = null }); + Publish(proxy, new FixturesEndpointModel + { + fixture = new fixture { id = "od:match:1", name = "Secret vs Liquid" } + }); + + Assert.False(escaped, EscapeMessage); + Assert.NotNull(cache.PeekMatch(MatchId)); + } + + // One malformed item must not take its siblings with it: the response-level + // guard would otherwise abort the whole batch. + [Fact] + public void PlayerCacheCachesTheGoodItemsInABatchWithABadId() + { + var (api, proxy) = NewApi(); + using var cache = new PlayerCache(api); + + Publish(proxy, Profile( + new player_profilePlayer { id = "bogus", name = "Malformed" }, + new player_profilePlayer { id = "od:player:1", name = "Dendi" })); + + Assert.NotNull(cache.GetPlayer(PlayerId, new[] { Culture })); + } + + [Fact] + public void CompetitorCacheCachesTheGoodItemsInABatchWithABadId() + { + var (api, proxy) = NewApi(); + using var cache = new CompetitorCache(api); + + Publish(proxy, new TournamentInfoModel + { + tournament = new tournamentExtended { id = "od:tournament:1", competitors = Array.Empty() }, + competitors = new[] + { + new team { id = "bogus", name = "Malformed" }, + new team { id = "od:competitor:1", name = "Team Secret" } + } + }); + + Assert.NotNull(cache.GetCompetitor(CompetitorId, new[] { Culture })); + } + + [Fact] + public void SportDataCacheCachesTheGoodItemsInABatchWithABadId() + { + var (api, proxy) = NewApi(); + using var cache = new SportDataCache(api); + + Publish(proxy, new TournamentScheduleModel + { + tournament = new[] + { + new tournamentExtended { id = "bogus", sport = new sport { id = "bogus", name = "Malformed" } }, + new tournamentExtended { id = "od:tournament:1", sport = Sport() } + } + }); + + Assert.NotNull(ReadSport(cache)); + } + + [Fact] + public void TournamentsCacheCachesTheGoodItemsInABatchWithABadId() + { + var (api, proxy) = NewApi(); + using var cache = new TournamentsCache(api); + + Publish(proxy, new TournamentsModel + { + tournaments = new List + { + new() { id = "bogus", name = "Malformed" }, + new() { id = "od:tournament:1", name = "The International" } + } + }); + + Assert.NotNull(cache.GetTournament(TournamentId, new[] { Culture })); + } + + [Fact] + public void MatchCacheCachesTheGoodItemsInABatchWithABadId() + { + var (api, proxy) = NewApi(); + using var cache = new MatchCache(api); + + Publish(proxy, new ScheduleEndpointModel + { + sport_event = new[] + { + new sportEvent { id = "bogus", name = "Malformed" }, + new sportEvent { id = "od:match:1", name = "Secret vs Liquid" } + } + }); + + Assert.NotNull(cache.PeekMatch(MatchId)); + } + + private const string EscapeMessage = + "a bad side-load response escaped into the API caller, which sees it as a failed request"; + + private static LocalizedSport ReadSport(SportDataCache cache) => + cache.GetSport(SportId, new[] { Culture }).ConfigureAwait(false).GetAwaiter().GetResult(); + + private static (IApiClient Api, PublishOnlyApiClientProxy Proxy) NewApi() + { + var api = DispatchProxy.Create(); + return (api, (PublishOnlyApiClientProxy)api); + } + + private static bool Publish(PublishOnlyApiClientProxy proxy, object data) + { + try + { + proxy.Publish(data, Culture); + return false; + } + catch + { + return true; + } + } + + private static sport Sport() => new() { id = "od:sport:1", name = "Dota 2" }; + + private static TournamentInfoModel TournamentInfo() => + new() + { + tournament = new tournamentExtended + { + id = "od:tournament:1", + name = "The International", + sport = Sport(), + competitors = Array.Empty() + } + }; + + private static competitorProfileEndpoint Profile(params player_profilePlayer[] players) => + new() + { + competitor = new teamExtended { id = "od:competitor:1", name = "Team Secret" }, + players = new List(players) + }; + + // Every direct API call fails, so a cache hit can only be a side-load. + public class PublishOnlyApiClientProxy : DispatchProxy + { + private readonly Subject> _responses = new(); + + public void Publish(object data, CultureInfo culture) => + _responses.OnNext(RequestResult.Success( + data, + HttpStatusCode.OK, + string.Empty, + culture: culture)); + + protected override object Invoke(MethodInfo targetMethod, object[] args) + { + if (targetMethod.Name == nameof(IApiClient.SubscribeForClass)) + return _responses.OfType>(); + + throw new InvalidOperationException($"API unavailable: {targetMethod.Name}"); + } + } +} diff --git a/src/Oddin.OddsFeedSdk/Oddin.OddsFeedSdk.Tests/API/CacheSideLoadDeadlockTests.cs b/src/Oddin.OddsFeedSdk/Oddin.OddsFeedSdk.Tests/API/CacheSideLoadDeadlockTests.cs new file mode 100644 index 0000000..a8ea0e1 --- /dev/null +++ b/src/Oddin.OddsFeedSdk/Oddin.OddsFeedSdk.Tests/API/CacheSideLoadDeadlockTests.cs @@ -0,0 +1,323 @@ +using System; +using System.Collections.Generic; +using System.Globalization; +using System.Linq; +using System.Net; +using System.Reactive.Linq; +using System.Reactive.Subjects; +using System.Reflection; +using System.Threading; +using System.Threading.Tasks; +using Oddin.OddsFeedSdk.API; +using Oddin.OddsFeedSdk.API.Abstractions; +using Oddin.OddsFeedSdk.API.Models; +using Oddin.OddsFeedSdk.Common; +using Xunit; + +namespace Oddin.OddsFeedSdk.Tests.API; + +[CollectionDefinition(Name, DisableParallelization = true)] +public sealed class CacheSideLoadCollection +{ + public const string Name = "Cache side-load"; +} + +// A response is published on the calling thread, so a sibling cache's observer +// takes a second semaphore there. Sport and tournament do it in opposite orders. +[Collection(CacheSideLoadCollection.Name)] +public class CacheSideLoadDeadlockTests +{ + private static readonly TimeSpan JoinTimeout = TimeSpan.FromSeconds(15); + private static readonly TimeSpan BarrierTimeout = TimeSpan.FromSeconds(5); + private static readonly CultureInfo Culture = new("en"); + + private static URN SportId => new("od:sport:1"); + private static URN TournamentId => new("od:tournament:1"); + private static URN MatchId => new("od:match:1"); + private static URN CompetitorId => new("od:competitor:1"); + private static URN PlayerId => new("od:player:1"); + + [Fact] + public void ConcurrentColdSportAndTournamentLoadsDoNotDeadlock() + { + var api = DispatchProxy.Create(); + var proxy = (SideLoadApiClientProxy)api; + proxy.ArmOverlapBarrier(BarrierTimeout); + + using var sportCache = new SportDataCache(api); + using var tournamentsCache = new TournamentsCache(api); + + var sportLoad = Worker.Start( + "sport-load", + () => sportCache.GetSportTournaments(SportId, Culture)); + var tournamentLoad = Worker.Start( + "tournament-load", + () => tournamentsCache.GetTournament(TournamentId, new[] { Culture })); + + var sportFinished = sportLoad.Join(JoinTimeout); + var tournamentFinished = tournamentLoad.Join(JoinTimeout); + + // Overlap first: without it a green run proves nothing. + Assert.True( + proxy.OverlapAchieved, + "the two loads never sat inside their API calls at the same time, so this run proved nothing"); + Assert.True(sportFinished, "sport load never finished — the two caches deadlocked"); + Assert.True(tournamentFinished, "tournament load never finished — the two caches deadlocked"); + + sportLoad.Rethrow(); + tournamentLoad.Rethrow(); + } + + [Theory] + [InlineData(CacheUnderTest.SportData)] + [InlineData(CacheUnderTest.Tournaments)] + [InlineData(CacheUnderTest.Competitor)] + [InlineData(CacheUnderTest.Player)] + [InlineData(CacheUnderTest.Match)] + [InlineData(CacheUnderTest.MatchStatus)] + public void CacheDoesNotHoldItsSemaphoreWhileTheApiCallIsInFlight(CacheUnderTest which) + { + var api = DispatchProxy.Create(); + var proxy = (SideLoadApiClientProxy)api; + + var (cache, exercise) = Build(which, api); + + using (cache as IDisposable) + { + bool? freeDuringCall = null; + proxy.WhileServing = () => freeDuringCall ??= SemaphoreIsFree(cache); + + exercise(); + + Assert.True(freeDuringCall.HasValue, $"{which} never made an API call, so this run proved nothing"); + Assert.True( + freeDuringCall.Value, + $"{which} held its semaphore across the API call — a sibling cache's observer " + + "runs on this thread and can be left waiting for it"); + } + } + + private static (object Cache, Action Exercise) Build(CacheUnderTest which, IApiClient api) + { + switch (which) + { + case CacheUnderTest.SportData: + var sportData = new SportDataCache(api); + return (sportData, () => sportData.GetSportTournaments(SportId, Culture)); + case CacheUnderTest.Tournaments: + var tournaments = new TournamentsCache(api); + return (tournaments, () => tournaments.GetTournament(TournamentId, new[] { Culture })); + case CacheUnderTest.Competitor: + var competitor = new CompetitorCache(api); + return (competitor, () => competitor.GetCompetitor(CompetitorId, new[] { Culture })); + case CacheUnderTest.Player: + var player = new PlayerCache(api); + return (player, () => player.GetPlayer(PlayerId, new[] { Culture })); + case CacheUnderTest.Match: + var match = new MatchCache(api); + return (match, () => match.GetMatch(MatchId, new[] { Culture })); + case CacheUnderTest.MatchStatus: + var matchStatus = new MatchStatusCache(api); + return (matchStatus, () => matchStatus.GetMatchStatus(MatchId)); + default: + throw new ArgumentOutOfRangeException(nameof(which)); + } + } + + // GetSports/GetSport share an async loader that was changed separately; the + // theory above only reaches the synchronous GetSportTournaments. + [Fact] + public async Task SportCacheDoesNotHoldItsSemaphoreDuringTheAsyncSportsCall() + { + var api = DispatchProxy.Create(); + var proxy = (SideLoadApiClientProxy)api; + + using var cache = new SportDataCache(api); + + bool? freeDuringCall = null; + proxy.WhileServing = () => freeDuringCall ??= SemaphoreIsFree(cache); + + var sports = await cache.GetSports(new[] { Culture }); + + Assert.True(freeDuringCall.HasValue, "GetSports never made an API call"); + Assert.True( + freeDuringCall.Value, + "SportDataCache held its semaphore across the async GetSports call"); + Assert.Contains(SportId, sports); + } + + // Semaphore(1,1) has no owning thread, so this reports the permit state from anywhere. + private static bool SemaphoreIsFree(object cache) + { + var field = cache.GetType().GetField("_semaphore", BindingFlags.Instance | BindingFlags.NonPublic); + Assert.NotNull(field); + + var semaphore = (Semaphore)field.GetValue(cache); + if (semaphore.WaitOne(TimeSpan.Zero) == false) + return false; + + semaphore.Release(); + return true; + } + + public enum CacheUnderTest + { + SportData, + Tournaments, + Competitor, + Player, + Match, + MatchStatus + } + + // Not the pool: on a failing run both bodies block forever. + private sealed class Worker + { + private readonly Thread _thread; + private Exception _error; + + private Worker(string name, Action body) + { + _thread = new Thread(() => + { + try + { + body(); + } + catch (Exception e) + { + _error = e; + } + }) + { + IsBackground = true, + Name = name + }; + } + + internal static Worker Start(string name, Action body) + { + var worker = new Worker(name, body); + worker._thread.Start(); + return worker; + } + + internal bool Join(TimeSpan timeout) => _thread.Join(timeout); + + // After the deadlock assertions, or it hides which failure happened. + internal void Rethrow() + { + if (_error is not null) + throw new Xunit.Sdk.XunitException($"{_thread.Name} threw: {_error}"); + } + } + + // Mirrors RestClient: publishes synchronously on the calling thread. + public class SideLoadApiClientProxy : DispatchProxy + { + private readonly Subject> _responses = new(); + private Barrier _overlap; + private TimeSpan _overlapTimeout; + private int _overlapReached; + + public Action WhileServing { get; set; } + + public bool OverlapAchieved => Volatile.Read(ref _overlapReached) >= 2; + + public void ArmOverlapBarrier(TimeSpan timeout) + { + _overlap = new Barrier(2); + _overlapTimeout = timeout; + } + + protected override object Invoke(MethodInfo targetMethod, object[] args) + { + switch (targetMethod.Name) + { + case nameof(IApiClient.SubscribeForClass): + return _responses.OfType>(); + case nameof(IApiClient.GetSports): + return Task.FromResult(Serve(SportsModel(), (CultureInfo)args[0])); + case nameof(IApiClient.GetTournaments): + return Serve(Tournaments(), (CultureInfo)args[1]); + case nameof(IApiClient.GetTournament): + return Serve(TournamentInfo(), (CultureInfo)args[1]); + case nameof(IApiClient.GetCompetitorProfileWithPlayers): + return Serve(CompetitorProfile(), (CultureInfo)args[1]); + case nameof(IApiClient.GetPlayerProfile): + return Serve(Player(), (CultureInfo)args[1]); + case nameof(IApiClient.GetMatchSummary): + return Serve(MatchSummary(), (CultureInfo)args[1]); + default: + throw new NotSupportedException($"Unexpected API call: {targetMethod.Name}"); + } + } + + private T Serve(T data, CultureInfo culture) + where T : class + { + WhileServing?.Invoke(); + + // Timed, so "never overlapped" fails as itself rather than hanging. + if (_overlap is not null && _overlap.SignalAndWait(_overlapTimeout)) + Interlocked.Increment(ref _overlapReached); + + _responses.OnNext(RequestResult.Success( + data, + HttpStatusCode.OK, + string.Empty, + culture: culture)); + + return data; + } + + // Only the fields the observers and loaders read. + private static sport Sport() => new() { id = "od:sport:1", name = "Dota 2" }; + + private static SportsModel SportsModel() => + new() { sport = new[] { new sportExtended { id = "od:sport:1", name = "Dota 2" } } }; + + private static TournamentsModel Tournaments() => + new() + { + sport = Sport(), + tournaments = new List + { + new() { id = "od:tournament:1", name = "The International", sport = Sport() } + } + }; + + private static TournamentInfoModel TournamentInfo() => + new() + { + tournament = new tournamentExtended + { + id = "od:tournament:1", + name = "The International", + sport = Sport(), + competitors = Array.Empty() + } + }; + + private static competitorProfileEndpoint CompetitorProfile() => + new() + { + competitor = new teamExtended { id = "od:competitor:1", name = "Team Secret" }, + players = new List() + }; + + private static player_profilePlayer Player() => + new() { id = "od:player:1", name = "Dendi", full_name = "Danil Ishutin" }; + + private static MatchSummaryModel MatchSummary() => + new() + { + sport_event = new sportEvent + { + id = "od:match:1", + name = "Secret vs Liquid", + tournament = new tournament { id = "od:tournament:1", sport = Sport() } + } + }; + } +} diff --git a/src/Oddin.OddsFeedSdk/Oddin.OddsFeedSdk.Tests/API/SideLoadDeliveryTests.cs b/src/Oddin.OddsFeedSdk/Oddin.OddsFeedSdk.Tests/API/SideLoadDeliveryTests.cs new file mode 100644 index 0000000..fd429ee --- /dev/null +++ b/src/Oddin.OddsFeedSdk/Oddin.OddsFeedSdk.Tests/API/SideLoadDeliveryTests.cs @@ -0,0 +1,208 @@ +using System; +using System.Collections.Generic; +using System.Globalization; +using System.Net; +using System.Net.Http; +using System.Net.Sockets; +using System.Reactive.Linq; +using System.Reflection; +using System.Text; +using System.Threading; +using System.Threading.Tasks; +using Oddin.OddsFeedSdk.API; +using Oddin.OddsFeedSdk.API.Abstractions; +using Oddin.OddsFeedSdk.API.Models; +using Oddin.OddsFeedSdk.Common; +using Oddin.OddsFeedSdk.Configuration.Abstractions; +using Xunit; + +namespace Oddin.OddsFeedSdk.Tests.API; + +// Side-load delivery is inline on purpose. GetMatchStatus, LoadFixture and the +// schedule warm-up all read the cache straight after their API call returns, so +// moving delivery to a scheduler breaks them silently. +[Collection(CacheSideLoadCollection.Name)] +public class SideLoadDeliveryTests +{ + private static readonly CultureInfo Culture = new("en"); + + [Fact] + public void RestClientDeliversResponsesOnTheCallingThread() + { + using var server = StubApiServer.Start(""); + using var restClient = new RestClient(new StubConfiguration(server.Authority)); + + var observerThread = 0; + using var subscription = restClient + .SubscribeForClass>() + .Subscribe(_ => observerThread = Environment.CurrentManagedThreadId); + + restClient.SendRequest("v1/sports/en/sports", HttpMethod.Get, Culture); + + Assert.True(observerThread != 0, "the response was never published"); + Assert.Equal(Environment.CurrentManagedThreadId, observerThread); + } + + [Fact] + public void GetMatchStatusReturnsTheStatusItJustFetched() + { + var api = DispatchProxy.Create(); + using var cache = new MatchStatusCache(api); + + Assert.NotNull(cache.GetMatchStatus(new URN("od:match:1"))); + } + + [Fact] + public void LoadFixtureLeavesTheMatchReadableByPeek() + { + var api = DispatchProxy.Create(); + using var cache = new MatchCache(api); + + var id = new URN("od:match:1"); + cache.LoadFixture(id, Culture); + + Assert.NotNull(cache.PeekMatch(id)); + } + + private sealed class StubApiServer : IDisposable + { + private readonly HttpListener _listener; + private readonly string _payload; + + private StubApiServer(HttpListener listener, string authority, string payload) + { + _listener = listener; + _payload = payload; + Authority = authority; + } + + public string Authority { get; } + + public static StubApiServer Start(string payload) + { + var port = FreePort(); + var listener = new HttpListener(); + listener.Prefixes.Add($"http://localhost:{port}/"); + listener.Start(); + + var server = new StubApiServer(listener, $"localhost:{port}", payload); + Task.Run(server.Serve); + return server; + } + + private async Task Serve() + { + while (_listener.IsListening) + { + HttpListenerContext context; + try + { + context = await _listener.GetContextAsync(); + } + catch (Exception) + { + return; + } + + var body = Encoding.UTF8.GetBytes(_payload); + context.Response.ContentType = "application/xml"; + context.Response.ContentLength64 = body.Length; + await context.Response.OutputStream.WriteAsync(body); + context.Response.Close(); + } + } + + private static int FreePort() + { + var probe = new TcpListener(IPAddress.Loopback, 0); + probe.Start(); + var port = ((IPEndPoint)probe.LocalEndpoint).Port; + probe.Stop(); + return port; + } + + public void Dispose() => _listener.Close(); + } + + private sealed class StubConfiguration : IFeedConfiguration + { + public StubConfiguration(string apiHost) => ApiHost = apiHost; + + public string AccessToken => "token"; + public CultureInfo DefaultLocale => Culture; + public int MaxInactivitySeconds => 20; + public int MaxRecoveryExecutionMinutes => 15; + public int? NodeId => null; + public ExceptionHandlingStrategy ExceptionHandlingStrategy => ExceptionHandlingStrategy.CATCH; + public string Host => "localhost"; + public int Port => 5672; + public bool UseSsl => false; + public string ApiHost { get; } + public bool UseApiSsl => false; + public int HttpClientTimeout => 30; + public int InitialSnapshotTimeInMinutes => 0; + } + + // Publishes synchronously before returning, as RestClient does. + public class SummaryApiClientProxy : DispatchProxy + { + private readonly System.Reactive.Subjects.Subject> _responses = new(); + + protected override object Invoke(MethodInfo targetMethod, object[] args) + { + switch (targetMethod.Name) + { + case nameof(IApiClient.SubscribeForClass): + return _responses.OfType>(); + case nameof(IApiClient.GetMatchSummary): + return Publish(Summary()); + case nameof(IApiClient.GetFixture): + return Publish(Fixture()); + default: + throw new NotSupportedException($"Unexpected API call: {targetMethod.Name}"); + } + } + + private T Publish(T data) + where T : class + { + _responses.OnNext(RequestResult.Success( + data, + HttpStatusCode.OK, + string.Empty, + culture: Culture)); + + return data; + } + + private static tournament Tournament() => + new() + { + id = "od:tournament:1", + sport = new sport { id = "od:sport:1", name = "Dota 2" } + }; + + private static MatchSummaryModel Summary() => + new() + { + sport_event = new sportEvent + { + id = "od:match:1", + name = "Secret vs Liquid", + tournament = Tournament() + }, + sport_event_status = new sportEventStatus { status = "live", match_status_code = 1 } + }; + + private static FixturesEndpointModel Fixture() => + new() + { + fixture = new fixture + { + id = "od:match:1", + name = "Secret vs Liquid", + tournament = Tournament() + } + }; + } +} diff --git a/src/Oddin.OddsFeedSdk/Oddin.OddsFeedSdk.Tests/API/SportDataCacheTournamentIdsTests.cs b/src/Oddin.OddsFeedSdk/Oddin.OddsFeedSdk.Tests/API/SportDataCacheTournamentIdsTests.cs new file mode 100644 index 0000000..cec7623 --- /dev/null +++ b/src/Oddin.OddsFeedSdk/Oddin.OddsFeedSdk.Tests/API/SportDataCacheTournamentIdsTests.cs @@ -0,0 +1,169 @@ +using System; +using System.Collections.Generic; +using System.Globalization; +using System.Linq; +using System.Net; +using System.Reactive.Linq; +using System.Reactive.Subjects; +using System.Reflection; +using System.Threading.Tasks; +using Oddin.OddsFeedSdk.API; +using Oddin.OddsFeedSdk.API.Abstractions; +using Oddin.OddsFeedSdk.API.Entities; +using Oddin.OddsFeedSdk.API.Models; +using Oddin.OddsFeedSdk.Common; +using Xunit; + +namespace Oddin.OddsFeedSdk.Tests.API; + +// The sport entry is NotRemovable with no expiry, so anything put in TournamentIds +// lives for the process and ISport.Tournaments hands it straight to the client. +[Collection(CacheSideLoadCollection.Name)] +public class SportDataCacheTournamentIdsTests +{ + private static readonly CultureInfo Culture = new("en"); + + private static URN SportId => new("od:sport:1"); + + [Fact] + public async Task RepeatedSideLoadsKeepOneEntryPerTournament() + { + var api = DispatchProxy.Create(); + var proxy = (SportsApiClientProxy)api; + using var cache = new SportDataCache(api); + + proxy.Publish(TournamentInfo(), Culture); + proxy.Publish(TournamentInfo(), Culture); + + var sport = await cache.GetSport(SportId, new[] { Culture }); + + Assert.NotNull(sport.TournamentIds); + Assert.Equal(1, sport.TournamentIds.Count); + Assert.Contains(new URN("od:tournament:1"), sport.TournamentIds); + } + + [Fact] + public async Task GetSportTournamentsPutsTheIdsOnTheSport() + { + var api = DispatchProxy.Create(); + var proxy = (SportsApiClientProxy)api; + using var cache = new SportDataCache(api); + + var returned = cache.GetSportTournaments(SportId, Culture).ToList(); + + var sport = await cache.GetSport(SportId, new[] { Culture }); + + Assert.Equal(2, returned.Count); + Assert.NotNull(sport.TournamentIds); + Assert.Equal( + new[] { new URN("od:tournament:1"), new URN("od:tournament:2") }.OrderBy(u => u.Id), + sport.TournamentIds.OrderBy(u => u.Id)); + } + + // Entities are handed out by reference and read outside the semaphore, so a + // set that is already published must never be mutated again. + [Fact] + public async Task ASideLoadDoesNotMutateAnAlreadyPublishedTournamentIdSet() + { + var api = DispatchProxy.Create(); + var proxy = (SportsApiClientProxy)api; + using var cache = new SportDataCache(api); + + proxy.Publish(TournamentInfo(), Culture); + var published = (await cache.GetSport(SportId, new[] { Culture })).TournamentIds; + Assert.Equal(1, published.Count); + + proxy.Publish(TournamentInfo("od:tournament:2"), Culture); + + Assert.Equal(1, published.Count); + Assert.Equal(2, (await cache.GetSport(SportId, new[] { Culture })).TournamentIds.Count); + } + + // A client enumerating ISport.Tournaments must get the same order every call, + // not server order once and hash order afterwards. + [Fact] + public async Task TheSportKeepsServerOrderAcrossCalls() + { + var api = DispatchProxy.Create(); + var proxy = (SportsApiClientProxy)api; + proxy.TournamentIds = new[] { "od:tournament:9", "od:tournament:3", "od:tournament:7" }; + + using var cache = new SportDataCache(api); + + var firstCall = cache.GetSportTournaments(SportId, Culture).ToList(); + var laterCall = (await cache.GetSport(SportId, new[] { Culture })).TournamentIds.ToList(); + + Assert.Equal( + new[] { new URN("od:tournament:9"), new URN("od:tournament:3"), new URN("od:tournament:7") }, + firstCall); + Assert.Equal(firstCall, laterCall); + } + + [Fact] + public void GetSportTournamentsSkipsAMalformedIdAndKeepsTheRest() + { + var api = DispatchProxy.Create(); + var proxy = (SportsApiClientProxy)api; + proxy.TournamentIds = new[] { "bogus", "od:tournament:2" }; + + using var cache = new SportDataCache(api); + + var returned = cache.GetSportTournaments(SportId, Culture).ToList(); + + Assert.Equal(new[] { new URN("od:tournament:2") }, returned); + } + + private static sport Sport() => new() { id = "od:sport:1", name = "Dota 2" }; + + private static TournamentInfoModel TournamentInfo(string tournamentId = "od:tournament:1") => + new() + { + tournament = new tournamentExtended + { + id = tournamentId, + name = "The International", + sport = Sport(), + competitors = Array.Empty() + } + }; + + private static SportsModel Sports() => + new() { sport = new[] { new sportExtended { id = "od:sport:1", name = "Dota 2" } } }; + + public class SportsApiClientProxy : DispatchProxy + { + private readonly Subject> _responses = new(); + + public string[] TournamentIds { get; set; } = { "od:tournament:1", "od:tournament:2" }; + + public void Publish(object data, CultureInfo culture) => + _responses.OnNext(RequestResult.Success( + data, + HttpStatusCode.OK, + string.Empty, + culture: culture)); + + protected override object Invoke(MethodInfo targetMethod, object[] args) + { + switch (targetMethod.Name) + { + case nameof(IApiClient.SubscribeForClass): + return _responses.OfType>(); + case nameof(IApiClient.GetSports): + return Task.FromResult(Sports()); + case nameof(IApiClient.GetTournaments): + var tournaments = new TournamentsModel + { + sport = Sport(), + tournaments = TournamentIds + .Select(id => new tournament { id = id, sport = Sport() }) + .ToList() + }; + Publish(tournaments, (CultureInfo)args[1]); + return tournaments; + default: + throw new InvalidOperationException($"API unavailable: {targetMethod.Name}"); + } + } + } +} diff --git a/src/Oddin.OddsFeedSdk/Oddin.OddsFeedSdk/API/CompetitorCache.cs b/src/Oddin.OddsFeedSdk/Oddin.OddsFeedSdk/API/CompetitorCache.cs index 7cd34df..67538db 100644 --- a/src/Oddin.OddsFeedSdk/Oddin.OddsFeedSdk/API/CompetitorCache.cs +++ b/src/Oddin.OddsFeedSdk/Oddin.OddsFeedSdk/API/CompetitorCache.cs @@ -30,50 +30,41 @@ public CompetitorCache(IApiClient apiClient) _subscription = apiClient.SubscribeForClass>() .Subscribe(response => { - if (response.Culture is null || response.Data is null) - return; - - var competitors = response.Data switch + // An escape here kills the subscription for good. + try { - FixturesEndpointModel f => f.fixture.competitors.ToArray(), - MatchSummaryModel m => m.sport_event.competitors.ToArray(), - ScheduleEndpointModel s => s.sport_event.SelectMany(e => e.competitors).ToArray(), - TournamentScheduleModel t => t.tournament.SelectMany(e => e.competitors).ToArray(), - TournamentInfoModel t => t.competitors.ToArray(), - _ => new team[0] - }; - - if (competitors.Any()) + HandleResponse(response); + } + catch (Exception e) { - _semaphore.WaitOne(); - try - { - _log.LogDebug($"Updating Competitor cache from API: {response.Data.GetType()}"); - HandleTeamData(response.Culture, competitors); - } - finally - { - _semaphore.Release(); - } + _log.LogError(e, "Failed to side-load competitors"); } }); } - public LocalizedCompetitor GetCompetitor(URN id, IEnumerable cultures) + private void HandleResponse(IRequestResult response) { - _semaphore.WaitOne(); - try + if (response.Culture is null || response.Data is null) + return; + + var competitors = response.Data switch { - var alreadyExisting = _cache.Get(id.ToString()) as LocalizedCompetitor; - var localesExisting = alreadyExisting?.LoadedLocals ?? new List(); - var toFetch = cultures.Except(localesExisting); + FixturesEndpointModel f => f.fixture.competitors.ToArray(), + MatchSummaryModel m => m.sport_event.competitors.ToArray(), + ScheduleEndpointModel s => s.sport_event.SelectMany(e => e.competitors).ToArray(), + TournamentScheduleModel t => t.tournament.SelectMany(e => e.competitors).ToArray(), + TournamentInfoModel t => t.competitors.ToArray(), + _ => new team[0] + }; - if (toFetch.Any()) - { - LoadAndCacheItem(id, toFetch); - } + if (competitors.Any() == false) + return; - return _cache.Get(id.ToString()) as LocalizedCompetitor; + _semaphore.WaitOne(); + try + { + _log.LogDebug($"Updating Competitor cache from API: {response.Data.GetType()}"); + HandleTeamData(response.Culture, competitors); } finally { @@ -81,29 +72,30 @@ public LocalizedCompetitor GetCompetitor(URN id, IEnumerable cultur } } + public LocalizedCompetitor GetCompetitor(URN id, IEnumerable cultures) + { + var toFetch = MissingCultures(id, cultures); + if (toFetch.Count > 0) + LoadAndCacheItem(id, toFetch); + + return Read(id); + } + public string GetCompetitorIconPath(URN id, CultureInfo culture) { var competitor = _cache.Get(id.ToString()) as LocalizedCompetitor; if (competitor?.IconPathLoaded == true) return competitor.IconPath; - _semaphore.WaitOne(); - try - { - LoadAndCacheItem(id, new[] { culture }); - competitor = _cache.Get(id.ToString()) as LocalizedCompetitor; - return competitor?.IconPath; - } - finally - { - _semaphore.Release(); - } + LoadAndCacheItem(id, new[] { culture }); + return Read(id)?.IconPath; } public void ClearCacheItem(URN id) => _cache.Remove(id.ToString()); public void Dispose() => _subscription.Dispose(); + // Must not be called holding the semaphore: it locks itself, and Semaphore is not reentrant. public void LoadAndCacheItem(URN id, IEnumerable cultures) { foreach (var culture in cultures) @@ -111,6 +103,7 @@ public void LoadAndCacheItem(URN id, IEnumerable cultures) competitorProfileEndpoint data; try { + // Unlocked: publishing runs the sibling observers, which take their own locks. data = _apiClient.GetCompetitorProfileWithPlayers(id, culture); } catch (Exception e) @@ -119,6 +112,7 @@ public void LoadAndCacheItem(URN id, IEnumerable cultures) continue; } + _semaphore.WaitOne(); try { RefreshOrInsertItem(id, culture, data); @@ -127,6 +121,38 @@ public void LoadAndCacheItem(URN id, IEnumerable cultures) { _log.LogError($"Error while refreshing competitor {e}"); } + finally + { + _semaphore.Release(); + } + } + } + + private List MissingCultures(URN id, IEnumerable cultures) + { + _semaphore.WaitOne(); + try + { + var alreadyExisting = _cache.Get(id.ToString()) as LocalizedCompetitor; + var localesExisting = alreadyExisting?.LoadedLocals ?? new List(); + return cultures.Except(localesExisting).ToList(); + } + finally + { + _semaphore.Release(); + } + } + + private LocalizedCompetitor Read(URN id) + { + _semaphore.WaitOne(); + try + { + return _cache.Get(id.ToString()) as LocalizedCompetitor; + } + finally + { + _semaphore.Release(); } } @@ -217,8 +243,16 @@ private void HandleTeamData(CultureInfo culture, team[] teams) { foreach (var team in teams) { - var id = string.IsNullOrEmpty(team?.id) ? null : new URN(team.id); - RefreshOrInsertItem(id, culture, team); + // Per item, so one malformed id does not drop the rest of the response. + try + { + var id = string.IsNullOrEmpty(team?.id) ? null : new URN(team.id); + RefreshOrInsertItem(id, culture, team); + } + catch (Exception e) + { + _log.LogError($"Failed to refresh or insert competitor '{team?.id}': {e}"); + } } } } \ No newline at end of file diff --git a/src/Oddin.OddsFeedSdk/Oddin.OddsFeedSdk/API/MatchCache.cs b/src/Oddin.OddsFeedSdk/Oddin.OddsFeedSdk/API/MatchCache.cs index a38286d..8aeb351 100644 --- a/src/Oddin.OddsFeedSdk/Oddin.OddsFeedSdk/API/MatchCache.cs +++ b/src/Oddin.OddsFeedSdk/Oddin.OddsFeedSdk/API/MatchCache.cs @@ -34,53 +34,65 @@ public MatchCache(IApiClient apiClient) _subscription = apiClient.SubscribeForClass>() .Subscribe(response => { - if (response.Culture is null || response.Data is null) - return; - - if (response.Data is FixturesEndpointModel fixture) + // An escape here kills the subscription for good. + try { - _semaphore.WaitOne(); - try - { - _log.LogDebug($"Updating Match cache from API: {response.Data.GetType()}"); - HandleMatchData( - response.Culture, - new List { fixture.fixture }, - fromFixture: true, - fixture.fixture?.extra_info); - } - finally - { - _semaphore.Release(); - } - - return; + HandleResponse(response); } - - var matches = response.Data switch - { - ScheduleEndpointModel s => s.sport_event.ToList(), - TournamentScheduleModel t => t.sport_events.SelectMany(s => s).ToList(), - _ => new List() - }; - - - if (matches.Any()) + catch (Exception e) { - _semaphore.WaitOne(); - try - { - _log.LogDebug($"Updating Match cache from API: {response.Data.GetType()}"); - HandleMatchData(response.Culture, matches, fromFixture: false); - } - finally - { - _semaphore.Release(); - } + _log.LogError(e, "Failed to side-load matches"); } }); } + private void HandleResponse(IRequestResult response) + { + if (response.Culture is null || response.Data is null) + return; + + if (response.Data is FixturesEndpointModel fixture) + { + _semaphore.WaitOne(); + try + { + _log.LogDebug($"Updating Match cache from API: {response.Data.GetType()}"); + HandleMatchData( + response.Culture, + new List { fixture.fixture }, + fromFixture: true, + fixture.fixture?.extra_info); + } + finally + { + _semaphore.Release(); + } + + return; + } + + var matches = response.Data switch + { + ScheduleEndpointModel s => s.sport_event.ToList(), + TournamentScheduleModel t => t.sport_events.SelectMany(s => s).ToList(), + _ => new List() + }; + + if (matches.Any() == false) + return; + + _semaphore.WaitOne(); + try + { + _log.LogDebug($"Updating Match cache from API: {response.Data.GetType()}"); + HandleMatchData(response.Culture, matches, fromFixture: false); + } + finally + { + _semaphore.Release(); + } + } + public void OnFeedMessageReceived(fixture_change e) { var id = string.IsNullOrEmpty(e?.event_id) @@ -95,17 +107,34 @@ public void OnFeedMessageReceived(fixture_change e) } public LocalizedMatch GetMatch(URN id, IEnumerable cultures) + { + var culturesToLoad = MissingCultures(id, cultures); + if (culturesToLoad.Count > 0) + LoadAndCacheItem(id, culturesToLoad); + + return Read(id); + } + + private List MissingCultures(URN id, IEnumerable cultures) { _semaphore.WaitOne(); try { var localizedMatch = _cache.Get(id.ToString()) as LocalizedMatch; var localizedAlready = localizedMatch?.LoadedLocals ?? new List(); + return cultures.Except(localizedAlready).ToList(); + } + finally + { + _semaphore.Release(); + } + } - var culturesToLoad = cultures.Except(localizedAlready); - if (culturesToLoad.Any()) - LoadAndCacheItem(id, culturesToLoad); - + private LocalizedMatch Read(URN id) + { + _semaphore.WaitOne(); + try + { return _cache.Get(id.ToString()) as LocalizedMatch; } finally @@ -139,6 +168,7 @@ private void LoadAndCacheItem(URN id, IEnumerable cultures) MatchSummaryModel matchData; try { + // Unlocked: publishing runs the sibling observers, which take their own locks. matchData = _apiClient.GetMatchSummary(id, culture); } catch (Exception e) @@ -147,6 +177,7 @@ private void LoadAndCacheItem(URN id, IEnumerable cultures) continue; } + _semaphore.WaitOne(); try { RefreshOrInsertItem(id, culture, matchData.sport_event, fromFixture: false); @@ -155,6 +186,10 @@ private void LoadAndCacheItem(URN id, IEnumerable cultures) { _log.LogError($"Failed to refresh or load match {culture.TwoLetterISOLanguageName}: {e}"); } + finally + { + _semaphore.Release(); + } } } @@ -194,11 +229,10 @@ private void RefreshOrInsertItem( } else { // A sport_format value a newer backend introduced that this SDK version does not - // recognise. Do not throw: this runs inside the Rx subscription arms, which have no - // catch, so the exception would escape OnNext and starve every cache that subscribed - // after MatchCache. Report Unknown, but leave hasSportFormat false so the insert - // branch surfaces it honestly while the update branch preserves a known cached value - // rather than demoting it. + // recognise. Do not throw: the nearest catch is the per-item one in HandleMatchData, + // so an unknown format would cost this whole match. Report Unknown, but leave + // hasSportFormat false so the insert branch surfaces it honestly while the update + // branch preserves a known cached value rather than demoting it. _log.LogWarning($"Unknown sport format '{sportFormatValue}' for match '{id}', treating as Unknown."); sportFormat = SportFormat.Unknown; } @@ -278,7 +312,7 @@ private void HandleMatchData( // many sportEvents, and both the URN construction below and RefreshOrInsertItem can throw // on a malformed server value (new URN(...) rejects anything that isn't three // colon-separated parts with a positive numeric id). These run inside the Rx subscription - // arms, which have no catch, so an escape would starve every later subscriber. Isolating + // arms, whose catch would drop the rest of the response with it. Isolating // per item keeps one bad match from dropping its siblings, matching LoadAndCacheItem's // log-and-continue contract. try diff --git a/src/Oddin.OddsFeedSdk/Oddin.OddsFeedSdk/API/MatchStatusCache.cs b/src/Oddin.OddsFeedSdk/Oddin.OddsFeedSdk/API/MatchStatusCache.cs index 25e00fc..368c887 100644 --- a/src/Oddin.OddsFeedSdk/Oddin.OddsFeedSdk/API/MatchStatusCache.cs +++ b/src/Oddin.OddsFeedSdk/Oddin.OddsFeedSdk/API/MatchStatusCache.cs @@ -31,24 +31,37 @@ public MatchStatusCache(IApiClient apiClient) _subscription = apiClient.SubscribeForClass>() .Subscribe(response => { - if (response.Data is not MatchSummaryModel summary) - return; - - var id = string.IsNullOrEmpty(summary.sport_event?.id) ? null : new URN(summary.sport_event.id); - - _semaphore.WaitOne(); + // An escape here kills the subscription for good. try { - _log.LogDebug($"Updating Match Status cache from API: {response.Data.GetType()}"); - RefreshOrInsertApiItem(id, summary.sport_event_status); + HandleResponse(response); } - finally + catch (Exception e) { - _semaphore.Release(); + _log.LogError(e, "Failed to side-load match status"); } }); } + private void HandleResponse(IRequestResult response) + { + if (response.Data is not MatchSummaryModel summary) + return; + + var id = string.IsNullOrEmpty(summary.sport_event?.id) ? null : new URN(summary.sport_event.id); + + _semaphore.WaitOne(); + try + { + _log.LogDebug($"Updating Match Status cache from API: {response.Data.GetType()}"); + RefreshOrInsertApiItem(id, summary.sport_event_status); + } + finally + { + _semaphore.Release(); + } + } + public void OnFeedMessageReceived(odds_change e) { if (e.sport_event_status == null) diff --git a/src/Oddin.OddsFeedSdk/Oddin.OddsFeedSdk/API/PlayerCache.cs b/src/Oddin.OddsFeedSdk/Oddin.OddsFeedSdk/API/PlayerCache.cs index 197396f..c1765a1 100644 --- a/src/Oddin.OddsFeedSdk/Oddin.OddsFeedSdk/API/PlayerCache.cs +++ b/src/Oddin.OddsFeedSdk/Oddin.OddsFeedSdk/API/PlayerCache.cs @@ -29,45 +29,73 @@ public PlayerCache(IApiClient apiClient) _subscription = apiClient.SubscribeForClass>() .Subscribe(response => { - if (response.Culture is null || response.Data is null) - return; - - var players = response.Data switch + // An escape here kills the subscription for good. + try { - competitorProfileEndpoint f => f.players.ToArray(), - _ => Array.Empty() - }; - - if (players.Any()) + HandleResponse(response); + } + catch (Exception e) { - _semaphore.WaitOne(); - try - { - _log.LogDebug($"Updating Player cache from API: {response.Data.GetType()}"); - HandlePlayersData(response.Culture, players); - } - finally - { - _semaphore.Release(); - } + _log.LogError(e, "Failed to side-load players"); } }); } + private void HandleResponse(IRequestResult response) + { + if (response.Culture is null || response.Data is null) + return; + + var players = response.Data switch + { + competitorProfileEndpoint f => f.players.ToArray(), + _ => Array.Empty() + }; + + if (players.Any() == false) + return; + + _semaphore.WaitOne(); + try + { + _log.LogDebug($"Updating Player cache from API: {response.Data.GetType()}"); + HandlePlayersData(response.Culture, players); + } + finally + { + _semaphore.Release(); + } + } + public LocalizedPlayer GetPlayer(URN id, IEnumerable cultures) + { + var toFetch = MissingCultures(id, cultures); + if (toFetch.Count > 0) + LoadAndCacheItem(id, toFetch); + + return Read(id); + } + + private List MissingCultures(URN id, IEnumerable cultures) { _semaphore.WaitOne(); try { var alreadyExisting = _cache.Get(id.ToString()) as LocalizedPlayer; var localesExisting = alreadyExisting?.LoadedLocals ?? new List(); - var toFetch = cultures.Except(localesExisting); - - if (toFetch.Any()) - { - LoadAndCacheItem(id, toFetch); - } + return cultures.Except(localesExisting).ToList(); + } + finally + { + _semaphore.Release(); + } + } + private LocalizedPlayer Read(URN id) + { + _semaphore.WaitOne(); + try + { return _cache.Get(id.ToString()) as LocalizedPlayer; } finally @@ -87,6 +115,7 @@ private void LoadAndCacheItem(URN id, IEnumerable cultures) player_profilePlayer player; try { + // Unlocked: publishing runs the sibling observers, which take their own locks. player = _apiClient.GetPlayerProfile(id, culture); } catch (Exception e) @@ -95,6 +124,7 @@ private void LoadAndCacheItem(URN id, IEnumerable cultures) continue; } + _semaphore.WaitOne(); try { RefreshOrInsertItem(id, culture, player); @@ -103,6 +133,10 @@ private void LoadAndCacheItem(URN id, IEnumerable cultures) { _log.LogError($"Error while refreshing player {e}"); } + finally + { + _semaphore.Release(); + } } } @@ -129,8 +163,16 @@ private void HandlePlayersData(CultureInfo culture, player_profilePlayer[] playe { foreach (var player in players) { - var id = string.IsNullOrEmpty(player?.id) ? null : new URN(player.id); - RefreshOrInsertItem(id, culture, player); + // Per item, so one malformed id does not drop the rest of the response. + try + { + var id = string.IsNullOrEmpty(player?.id) ? null : new URN(player.id); + RefreshOrInsertItem(id, culture, player); + } + catch (Exception e) + { + _log.LogError($"Failed to refresh or insert player '{player?.id}': {e}"); + } } } } \ No newline at end of file diff --git a/src/Oddin.OddsFeedSdk/Oddin.OddsFeedSdk/API/SportDataCache.cs b/src/Oddin.OddsFeedSdk/Oddin.OddsFeedSdk/API/SportDataCache.cs index a8792c6..4a2603c 100644 --- a/src/Oddin.OddsFeedSdk/Oddin.OddsFeedSdk/API/SportDataCache.cs +++ b/src/Oddin.OddsFeedSdk/Oddin.OddsFeedSdk/API/SportDataCache.cs @@ -20,7 +20,8 @@ internal class SportDataCache : ISportDataCache private readonly IApiClient _apiClient; private readonly MemoryCache _cache = new(nameof(SportDataCache)); private readonly CacheItemPolicy _cachePolicy = new() { Priority = CacheItemPriority.NotRemovable }; - private readonly IList _loadedLocales = new List(); + // Set: the check and the add are in different critical sections. + private readonly HashSet _loadedLocales = new(); private readonly Semaphore _semaphore = new(1, 1); @@ -32,46 +33,38 @@ public SportDataCache(IApiClient apiClient) _subscription = apiClient.SubscribeForClass>() .Subscribe(response => { - if (response.Culture is null || response.Data is null) - return; - - var tournamentData = response.Data switch + // An escape here kills the subscription for good. + try { - TournamentScheduleModel t => t.tournament.ToDictionary(t => t.id, t => t.sport), - TournamentInfoModel t => new Dictionary { { t.tournament.id, t.tournament.sport } }, - _ => new Dictionary() - }; - - if (tournamentData.Any()) + HandleResponse(response); + } + catch (Exception e) { - _semaphore.WaitOne(); - try - { - _log.LogDebug($"Updating SportData cache from API: {response.Data.GetType()}"); - HandleTournamentData(response.Culture, tournamentData); - } - finally - { - _semaphore.Release(); - } + _log.LogError(e, "Failed to side-load sports"); } }); } - public async Task> GetSports(IEnumerable cultures) + private void HandleResponse(IRequestResult response) { + if (response.Culture is null || response.Data is null) + return; + + var tournamentData = response.Data switch + { + TournamentScheduleModel t => t.tournament.ToDictionary(t => t.id, t => t.sport), + TournamentInfoModel t => new Dictionary { { t.tournament.id, t.tournament.sport } }, + _ => new Dictionary() + }; + + if (tournamentData.Any() == false) + return; + _semaphore.WaitOne(); try { - var culturesToLoad = cultures.Except(_loadedLocales); - if (culturesToLoad.Any()) - await LoadAndCacheItem(culturesToLoad); - - return _cache.Select(item => - { - var sport = item.Value as LocalizedSport; - return sport?.Id; - }); + _log.LogDebug($"Updating SportData cache from API: {response.Data.GetType()}"); + HandleTournamentData(response.Culture, tournamentData); } finally { @@ -79,20 +72,20 @@ public async Task> GetSports(IEnumerable cultures) } } - public async Task GetSport(URN id, IEnumerable cultures) + public async Task> GetSports(IEnumerable cultures) { + var culturesToLoad = MissingLocales(cultures); + if (culturesToLoad.Count > 0) + await LoadAndCacheItem(culturesToLoad); + _semaphore.WaitOne(); try { - var localizedSport = _cache.Get(id.ToString()) as LocalizedSport; - var sportCultures = localizedSport?.LoadedLocals ?? new List(); - var toLoadCultures = cultures.Except(sportCultures); - if (toLoadCultures.Any()) + return _cache.Select(item => { - await LoadAndCacheItem(toLoadCultures); - } - - return _cache.Get(id.ToString()) as LocalizedSport; + var sport = item.Value as LocalizedSport; + return sport?.Id; + }).ToList(); } finally { @@ -100,36 +93,118 @@ public async Task GetSport(URN id, IEnumerable cult } } + public async Task GetSport(URN id, IEnumerable cultures) + { + var toLoadCultures = MissingCultures(id, cultures); + if (toLoadCultures.Count > 0) + await LoadAndCacheItem(toLoadCultures); + + return Read(id); + } + public IEnumerable GetSportTournaments(URN id, CultureInfo culture) { - _semaphore.WaitOne(); + TournamentsModel tournaments; try { - TournamentsModel tournaments; - try - { - tournaments = _apiClient.GetTournaments(id, culture); - } - catch (Exception e) - { - _log.LogError($"Error while fetching sport tournaments {culture.TwoLetterISOLanguageName}: {e}"); - return null; - } + // Unlocked: publishing runs the sibling observers, which take their own locks. + tournaments = _apiClient.GetTournaments(id, culture); + } + catch (Exception e) + { + _log.LogError($"Error while fetching sport tournaments {culture.TwoLetterISOLanguageName}: {e}"); + return null; + } + + var tournamentIds = new List(); - var tournamentIds = tournaments.tournaments.Select(t => string.IsNullOrEmpty(t?.id) ? null : new URN(t.id)); - foreach (var tournamentId in tournamentIds) + _semaphore.WaitOne(); + try + { + foreach (var tournament in tournaments.tournaments) { + // Per item: new URN throws on a malformed server id, and outside the + // try it would abort the whole loop. try { - RefreshOrInsertItem(id, culture, tournamentId: tournamentId); + if (string.IsNullOrEmpty(tournament?.id)) + continue; + + tournamentIds.Add(new URN(tournament.id)); } catch (Exception e) { - _log.LogError($"Failed to insert or refresh sport tournaments: {e}"); + _log.LogError($"Failed to read sport tournament '{tournament?.id}': {e}"); } } - return tournamentIds; + // Once, after the loop: a reader takes the reference without the + // semaphore, so it must never see a half-filled set. + RefreshOrInsertItem(id, culture, tournamentIds: tournamentIds); + } + finally + { + _semaphore.Release(); + } + + return tournamentIds; + } + + private List MissingLocales(IEnumerable cultures) + { + _semaphore.WaitOne(); + try + { + return cultures.Except(_loadedLocales).ToList(); + } + finally + { + _semaphore.Release(); + } + } + + private List MissingCultures(URN id, IEnumerable cultures) + { + _semaphore.WaitOne(); + try + { + var localizedSport = _cache.Get(id.ToString()) as LocalizedSport; + var sportCultures = localizedSport?.LoadedLocals ?? new List(); + return cultures.Except(sportCultures).ToList(); + } + finally + { + _semaphore.Release(); + } + } + + // Copy-on-write: the published entry is handed out by reference and read + // outside the semaphore, so a visible set is never mutated. Returns current + // untouched when there is nothing to add, so "unknown" never becomes "empty". + private static ICollection WithTournaments(ICollection current, ICollection tournamentIds) + { + if (tournamentIds is null || tournamentIds.Count == 0) + return current; + + // Ordered, not a set: ISport.Tournaments enumerates this, and a client would + // otherwise get server order on the first call and hash order afterwards. + var updated = current is null ? new List() : new List(current); + var known = new HashSet(updated); + foreach (var tournamentId in tournamentIds) + { + if (known.Add(tournamentId)) + updated.Add(tournamentId); + } + + return updated; + } + + private LocalizedSport Read(URN id) + { + _semaphore.WaitOne(); + try + { + return _cache.Get(id.ToString()) as LocalizedSport; } finally { @@ -141,20 +216,45 @@ public IEnumerable GetSportTournaments(URN id, CultureInfo culture) private void HandleTournamentData(CultureInfo culture, Dictionary tournamentData) { + // Grouped so each sport is published once, however many of its tournaments + // this response carries. + var bySport = new Dictionary Tournaments)>(); + foreach (var tournament in tournamentData) { - var tournamentId = string.IsNullOrEmpty(tournament.Key) ? null : new URN(tournament.Key); - var sportId = string.IsNullOrEmpty(tournament.Key) ? null : new URN(tournament.Value.id); + // Per item, so one malformed id does not drop the rest of the response. + try + { + if (string.IsNullOrEmpty(tournament.Key) || string.IsNullOrEmpty(tournament.Value?.id)) + continue; + + var tournamentId = new URN(tournament.Key); + var sportKey = new URN(tournament.Value.id).ToString(); - if (sportId is null) continue; + if (bySport.TryGetValue(sportKey, out var entry) == false) + entry = bySport[sportKey] = (tournament.Value, new List()); - RefreshOrInsertItem(sportId, culture, tournament.Value); - var sport = _cache.Get(sportId.ToString()) as LocalizedSport; - if (sport is not null) + entry.Tournaments.Add(tournamentId); + } + catch (Exception e) { - var sportTournaments = sport.TournamentIds ??= new List(); - sportTournaments.Add(tournamentId); - sport.TournamentIds = sportTournaments; + _log.LogError($"Failed to read sport for tournament '{tournament.Key}': {e}"); + } + } + + foreach (var entry in bySport) + { + try + { + RefreshOrInsertItem( + new URN(entry.Key), + culture, + entry.Value.Sport, + entry.Value.Tournaments); + } + catch (Exception e) + { + _log.LogError($"Failed to refresh or insert sport '{entry.Key}': {e}"); } } } @@ -166,6 +266,7 @@ private async Task LoadAndCacheItem(IEnumerable cultures) SportsModel sports; try { + // Unlocked: publishing runs the sibling observers, which take their own locks. sports = await _apiClient.GetSports(culture); } catch (Exception e) @@ -174,24 +275,36 @@ private async Task LoadAndCacheItem(IEnumerable cultures) continue; } - foreach (var sport in sports.sport) + _semaphore.WaitOne(); + try { - var id = string.IsNullOrEmpty(sport?.id) ? null : new URN(sport.id); - try - { - RefreshOrInsertItem(id, culture, sport); - } - catch (Exception e) + foreach (var sport in sports.sport) { - _log.LogError($"Failed to insert or refresh sport: {e}"); + var id = string.IsNullOrEmpty(sport?.id) ? null : new URN(sport.id); + try + { + RefreshOrInsertItem(id, culture, sport); + } + catch (Exception e) + { + _log.LogError($"Failed to insert or refresh sport: {e}"); + } } - } - _loadedLocales.Add(culture); + _loadedLocales.Add(culture); + } + finally + { + _semaphore.Release(); + } } } - private void RefreshOrInsertItem(URN id, CultureInfo culture, sport sport = null, URN tournamentId = null) + private void RefreshOrInsertItem( + URN id, + CultureInfo culture, + sport sport = null, + ICollection tournamentIds = null) { var localizedSportItem = _cache.Get(id.ToString()); @@ -205,8 +318,7 @@ private void RefreshOrInsertItem(URN id, CultureInfo culture, sport sport = null if (sport is sportExtended sportExtended) localizedSport.IconPath = sportExtended.icon_path; - if (tournamentId != null) - localizedSport.TournamentIds ??= new List(); + localizedSport.TournamentIds = WithTournaments(localizedSport.TournamentIds, tournamentIds); _cache.Set(id.ToString(), localizedSport, _cachePolicy); } diff --git a/src/Oddin.OddsFeedSdk/Oddin.OddsFeedSdk/API/TournamentsCache.cs b/src/Oddin.OddsFeedSdk/Oddin.OddsFeedSdk/API/TournamentsCache.cs index 4d7753d..0670e4e 100644 --- a/src/Oddin.OddsFeedSdk/Oddin.OddsFeedSdk/API/TournamentsCache.cs +++ b/src/Oddin.OddsFeedSdk/Oddin.OddsFeedSdk/API/TournamentsCache.cs @@ -30,37 +30,49 @@ public TournamentsCache(IApiClient apiClient) _subscription = apiClient.SubscribeForClass>() .Subscribe(response => { - if (response.Culture is null || response.Data is null) - return; - - var tournaments = response.Data switch + // An escape here kills the subscription for good. + try { - FixturesEndpointModel f => new[] { f.fixture.tournament }, - TournamentsModel t => t.tournaments?.ToArray() ?? Array.Empty(), - MatchSummaryModel m => new[] { m.sport_event.tournament }, - ScheduleEndpointModel s => s.sport_event.Select(t => t.tournament).ToArray(), - TournamentScheduleModel t => t.tournament.ToArray(), - SportTournamentsModel s => s.tournaments?.ToArray() ?? Array.Empty(), - _ => Array.Empty() - }; - - - if (tournaments.Any()) + HandleResponse(response); + } + catch (Exception e) { - _semaphore.WaitOne(); - try - { - _log.LogDebug($"Updating Tournament cache from API: {response.Data.GetType()}"); - HandleTournamentsData(response.Culture, tournaments); - } - finally - { - _semaphore.Release(); - } + _log.LogError(e, "Failed to side-load tournaments"); } }); } + private void HandleResponse(IRequestResult response) + { + if (response.Culture is null || response.Data is null) + return; + + var tournaments = response.Data switch + { + FixturesEndpointModel f => new[] { f.fixture.tournament }, + TournamentsModel t => t.tournaments?.ToArray() ?? Array.Empty(), + MatchSummaryModel m => new[] { m.sport_event.tournament }, + ScheduleEndpointModel s => s.sport_event.Select(t => t.tournament).ToArray(), + TournamentScheduleModel t => t.tournament.ToArray(), + SportTournamentsModel s => s.tournaments?.ToArray() ?? Array.Empty(), + _ => Array.Empty() + }; + + if (tournaments.Any() == false) + return; + + _semaphore.WaitOne(); + try + { + _log.LogDebug($"Updating Tournament cache from API: {response.Data.GetType()}"); + HandleTournamentsData(response.Culture, tournaments); + } + finally + { + _semaphore.Release(); + } + } + public void OnFeedMessageReceived(fixture_change e) { var id = string.IsNullOrEmpty(e?.event_id) ? null : new URN(e.event_id); @@ -74,37 +86,17 @@ public void OnFeedMessageReceived(fixture_change e) public LocalizedTournament GetTournament(URN id, IEnumerable cultures) { - _semaphore.WaitOne(); - try - { - var localizedTournament = _cache.Get(id.ToString()) as LocalizedTournament; - var localizedAlready = localizedTournament?.LoadedLocals ?? new List(); + var culturesToLoad = MissingCultures(id, cultures); + if (culturesToLoad.Count > 0) + LoadAndCacheItem(id, culturesToLoad); - var culturesToLoad = cultures.Except(localizedAlready); - if (culturesToLoad.Any()) - LoadAndCacheItem(id, culturesToLoad); - - return _cache.Get(id.ToString()) as LocalizedTournament; - } - finally - { - _semaphore.Release(); - } + return Read(id); } public IEnumerable GetTournamentCompetitors(URN id, CultureInfo culture) { - _semaphore.WaitOne(); - try - { - LoadAndCacheItem(id, new[] { culture }); - var tournament = _cache.Get(id.ToString()) as LocalizedTournament; - return tournament?.CompetitorIds; - } - finally - { - _semaphore.Release(); - } + LoadAndCacheItem(id, new[] { culture }); + return Read(id)?.CompetitorIds; } public void ClearCacheItem(URN id) => _cache.Remove(id.ToString()); @@ -115,19 +107,47 @@ private void HandleTournamentsData(CultureInfo culture, tournament[] tournaments { foreach (var tournament in tournaments) { - var id = string.IsNullOrEmpty(tournament?.id) ? null : new URN(tournament.id); - + // Per item, so one malformed id does not drop the rest of the response. try { + var id = string.IsNullOrEmpty(tournament?.id) ? null : new URN(tournament.id); RefreshOrInsertItem(id, culture, tournament); } catch (Exception ex) { - _log.LogError(ex, "Failed to refresh or load tournament"); + _log.LogError(ex, $"Failed to refresh or load tournament '{tournament?.id}'"); } } } + private List MissingCultures(URN id, IEnumerable cultures) + { + _semaphore.WaitOne(); + try + { + var localizedTournament = _cache.Get(id.ToString()) as LocalizedTournament; + var localizedAlready = localizedTournament?.LoadedLocals ?? new List(); + return cultures.Except(localizedAlready).ToList(); + } + finally + { + _semaphore.Release(); + } + } + + private LocalizedTournament Read(URN id) + { + _semaphore.WaitOne(); + try + { + return _cache.Get(id.ToString()) as LocalizedTournament; + } + finally + { + _semaphore.Release(); + } + } + private void LoadAndCacheItem(URN id, IEnumerable cultures) { foreach (var culture in cultures) @@ -135,6 +155,7 @@ private void LoadAndCacheItem(URN id, IEnumerable cultures) TournamentInfoModel tournamentData; try { + // Unlocked: publishing runs the sibling observers, which take their own locks. tournamentData = _apiClient.GetTournament(id, culture); } catch (Exception e) @@ -143,6 +164,7 @@ private void LoadAndCacheItem(URN id, IEnumerable cultures) continue; } + _semaphore.WaitOne(); try { RefreshOrInsertItem(id, culture, tournamentData.tournament); @@ -151,6 +173,10 @@ private void LoadAndCacheItem(URN id, IEnumerable cultures) { _log.LogError($"Failed to refresh or load tournament {culture.TwoLetterISOLanguageName}: {e}"); } + finally + { + _semaphore.Release(); + } } }