Convert security access to async (SaveAsync/GetSecurityAsync) end to end

The security get-or-create path was bridged to the async storage through a
blocking AsyncHelper.Run, turning SaveAsync into a sync-over-async call inside
the hot message-processing path. Propagate real async up the whole chain.

TraderHelper:
- Add GetOrCreateAsync that awaits ISecurityStorage.SaveAsync. It does not hold
  the storage Lock.Scope across the await (the lock is monitor based and thread
  affine); SaveAsync(forced: false) is an atomic add-if-absent, and the
  canonical stored instance is re-resolved to stay correct under concurrent
  creation.
- Mark the sync GetOrCreate(out bool) obsolete; it bridges to the async variant
  and has no internal callers left.

Connector / Connector_ProcessMessage:
- Add GetSecurityAsync/EnsureGetSecurityAsync and route the message-processing
  handlers (Security, Level1, News, Quotes, Trade, Transaction, PositionChange,
  Execution) through them; the already-async dispatcher awaits them. No blocking
  Run remains on this path.
- TryGetSecurity (entity cache snapshot delegate) becomes a pure LookupById: a
  snapshot lookup must not create or persist a security.

IConnector:
- Add GetSecurityAsync(SecurityId, CancellationToken) and mark the sync
  GetSecurity(SecurityId) obsolete. The sync method stays as a deprecated bridge
  for external callers only.

Migrate the in-repo callers (tests and the RemoteSource sample) to the async API.

Author: mikasoukhov

Algo/Connector.cs

@@ -955,13 +955,16 @@ private void ProcessTimeInterval(Message message)
 	}
 
 	/// <inheritdoc />
+	[Obsolete("Use GetSecurityAsync instead.")]
 	public Security GetSecurity(SecurityId securityId)
-		=> GetSecurity(securityId, s => false);
+		// Sync bridge over the async path kept only for the deprecated IConnector contract.
+		=> AsyncHelper.Run(() => GetSecurityAsync(securityId, default));
 
 	private Security TryGetSecurity(SecurityId? securityId)
-		=> securityId == null || securityId.Value == default ? null : GetSecurity(securityId.Value);
+		// Pure lookup used by the entity cache snapshots: it must not create/persist a security.
+		=> securityId == null || securityId.Value == default ? null : SecurityStorage.LookupById(securityId.Value);
 
-	private Security EnsureGetSecurity<TMessage>(TMessage message)
+	private async ValueTask<Security> EnsureGetSecurityAsync<TMessage>(TMessage message, CancellationToken cancellationToken)
 		where TMessage : ISecurityIdMessage, ISubscriptionIdMessage
 	{
 		var secId = message.SecurityId;
@@ -983,24 +986,29 @@ private Security EnsureGetSecurity<TMessage>(TMessage message)
 			secId = subscrSecId.Value;
 		}
 
-		return TryGetSecurity(secId) ?? throw new ArgumentOutOfRangeException(nameof(message), message, LocalizedStrings.SecurityNoFound.Put(secId));
+		return await GetSecurityAsync(secId, cancellationToken) ?? throw new ArgumentOutOfRangeException(nameof(message), message, LocalizedStrings.SecurityNoFound.Put(secId));
 	}
 
+	/// <inheritdoc />
+	public ValueTask<Security> GetSecurityAsync(SecurityId securityId, CancellationToken cancellationToken)
+		=> GetSecurityAsync(securityId, s => false, cancellationToken);
+
 	/// <summary>
 	/// To get the instrument by the code.
 	/// </summary>
 	/// <param name="id">Security ID.</param>
 	/// <param name="changeSecurity">The handler changing the instrument. It returns <see langword="true" /> if the instrument has been changed and the <see cref="SecurityReceived"/> should be called.</param>
+	/// <param name="cancellationToken"><see cref="CancellationToken"/></param>
 	/// <returns>Security.</returns>
-	private Security GetSecurity(SecurityId id, Func<Security, bool> changeSecurity)
+	private async ValueTask<Security> GetSecurityAsync(SecurityId id, Func<Security, bool> changeSecurity, CancellationToken cancellationToken)
 	{
 		if (id == default)
 			throw new ArgumentNullException(nameof(id));
 
 		if (changeSecurity == null)
 			throw new ArgumentNullException(nameof(changeSecurity));
 
-		var security = SecurityStorage.GetOrCreate(id, key =>
+		var (security, isNew) = await SecurityStorage.GetOrCreateAsync(id, key =>
 		{
 			var idInfo = SecurityIdGenerator.Split(key);
 
@@ -1013,7 +1021,7 @@ private Security GetSecurity(SecurityId id, Func<Security, bool> changeSecurity)
 				Code = code,
 				Board = board,
 			};
-		}, out var isNew);
+		}, cancellationToken);
 
 		if (_existingSecurities.TryAdd(security))
 			_added?.Invoke([security]);

Algo/Connector_ProcessMessage.cs

@@ -617,7 +617,7 @@ protected virtual async ValueTask OnProcessMessage(Message message, Cancellation
 					break;
 
 				case MessageTypes.QuoteChange:
-					ProcessQuotesMessage((QuoteChangeMessage)message);
+					await ProcessQuotesMessage((QuoteChangeMessage)message, cancellationToken);
 					break;
 
 				case MessageTypes.Board:
@@ -629,31 +629,31 @@ protected virtual async ValueTask OnProcessMessage(Message message, Cancellation
 					break;
 
 				case MessageTypes.Security:
-					ProcessSecurityMessage((SecurityMessage)message);
+					await ProcessSecurityMessage((SecurityMessage)message, cancellationToken);
 					break;
 
 				case MessageTypes.DataTypeInfo:
 					ProcessDataTypeInfoMessage((DataTypeInfoMessage)message);
 					break;
 
 				case MessageTypes.Level1Change:
-					ProcessLevel1ChangeMessage((Level1ChangeMessage)message);
+					await ProcessLevel1ChangeMessage((Level1ChangeMessage)message, cancellationToken);
 					break;
 
 				case MessageTypes.News:
-					ProcessNewsMessage((NewsMessage)message);
+					await ProcessNewsMessage((NewsMessage)message, cancellationToken);
 					break;
 
 				case MessageTypes.Execution:
-					ProcessExecutionMessage((ExecutionMessage)message);
+					await ProcessExecutionMessage((ExecutionMessage)message, cancellationToken);
 					break;
 
 				case MessageTypes.Portfolio:
 					ProcessPortfolioMessage((PortfolioMessage)message);
 					break;
 
 				case MessageTypes.PositionChange:
-					ProcessPositionChangeMessage((PositionChangeMessage)message);
+					await ProcessPositionChangeMessage((PositionChangeMessage)message, cancellationToken);
 					break;
 
 				//case MessageTypes.Time:
@@ -771,7 +771,7 @@ private async ValueTask ProcessSubscriptionFinishedMessage(SubscriptionFinishedM
 
 				await foreach (var secMsg in message.Body.ExtractSecuritiesAsync().WithCancellation(cancellationToken))
 				{
-					ProcessSecurityMessage(secMsg);
+					await ProcessSecurityMessage(secMsg, cancellationToken);
 					secMsgs.Add(secMsg);
 				}
 
@@ -923,16 +923,16 @@ private void ProcessBoardMessage(BoardMessage message)
 		RaiseReceived(board, subscriptions, BoardReceived);
 	}
 
-	private void ProcessSecurityMessage(SecurityMessage message)
+	private async ValueTask ProcessSecurityMessage(SecurityMessage message, CancellationToken cancellationToken)
 	{
-		var security = GetSecurity(message.SecurityId, s =>
+		var security = await GetSecurityAsync(message.SecurityId, s =>
 		{
 			if (!UpdateSecurityByDefinition)
 				return false;
 
 			s.ApplyChanges(message, ExchangeInfoProvider, OverrideSecurityData);
 			return true;
-		});
+		}, cancellationToken);
 
 		var subscriptions = _subscriptionManager.ProcessLookupResponse(message, security);
 		RaiseReceived(security, subscriptions, SecurityReceived);
@@ -946,7 +946,7 @@ private void ProcessDataTypeInfoMessage(DataTypeInfoMessage message)
 		RaiseReceived(dt, message, DataTypeReceived);
 	}
 
-	private void ProcessLevel1ChangeMessage(Level1ChangeMessage message)
+	private async ValueTask ProcessLevel1ChangeMessage(Level1ChangeMessage message, CancellationToken cancellationToken)
 	{
 		Security security = null;
 
@@ -955,7 +955,7 @@ private void ProcessLevel1ChangeMessage(Level1ChangeMessage message)
 			if (anyCanOnline != true)
 				return;
 
-			security = EnsureGetSecurity(message);
+			security = await EnsureGetSecurityAsync(message, cancellationToken);
 
 			if (_entityCache.HasLevel1Info(security))
 				return;
@@ -964,15 +964,15 @@ private void ProcessLevel1ChangeMessage(Level1ChangeMessage message)
 #pragma warning disable CS0618 // Type or member is obsolete
 		if (UpdateSecurityByLevel1)
 		{
-			security ??= EnsureGetSecurity(message);
+			security ??= await EnsureGetSecurityAsync(message, cancellationToken);
 
 			security.ApplyChanges(message);
 		}
 #pragma warning restore CS0618 // Type or member is obsolete
 
 		if (ValuesChanged is not null)
 		{
-			security ??= EnsureGetSecurity(message);
+			security ??= await EnsureGetSecurityAsync(message, cancellationToken);
 
 			var time = message.ServerTime;
 			var info = _entityCache.GetSecurityValues(security, time);
@@ -1069,7 +1069,7 @@ private void ProcessPortfolioMessage(PortfolioMessage message)
 		RaiseReceived(portfolio, message, PortfolioReceived);
 	}
 
-	private void ProcessPositionChangeMessage(PositionChangeMessage message)
+	private async ValueTask ProcessPositionChangeMessage(PositionChangeMessage message, CancellationToken cancellationToken)
 	{
 		if (!message.StrategyId.IsEmpty())
 			return;
@@ -1090,7 +1090,7 @@ private void ProcessPositionChangeMessage(PositionChangeMessage message)
 			RaiseReceived(portfolio, message, PortfolioReceived);
 		}
 
-		var security = EnsureGetSecurity(message);
+		var security = await EnsureGetSecurityAsync(message, cancellationToken);
 		portfolio = LookupByPortfolioName(message.PortfolioName);
 
 		var valueInLots = message.TryGetDecimal(PositionChangeTypes.CurrentValueInLots);
@@ -1112,17 +1112,17 @@ private void ProcessPositionChangeMessage(PositionChangeMessage message)
 		RaiseReceived(position, message, PositionReceived);
 	}
 
-	private void ProcessNewsMessage(NewsMessage message)
+	private async ValueTask ProcessNewsMessage(NewsMessage message, CancellationToken cancellationToken)
 	{
-		var security = message.SecurityId == null ? null : GetSecurity(message.SecurityId.Value);
+		var security = message.SecurityId == null ? null : await GetSecurityAsync(message.SecurityId.Value, cancellationToken);
 
 		var news = _entityCache.ProcessNewsMessage(security, message);
 
 		if (RaiseReceived(news.news, message, NewsReceived) == false)
 			return;
 	}
 
-	private void ProcessQuotesMessage(QuoteChangeMessage message)
+	private async ValueTask ProcessQuotesMessage(QuoteChangeMessage message, CancellationToken cancellationToken)
 	{
 		if (RaiseReceived(message, message, OrderBookReceived) != true)
 			return;
@@ -1141,7 +1141,7 @@ private void ProcessQuotesMessage(QuoteChangeMessage message)
 
 		if (ValuesChanged is not null && !fromLevel1 && !Adapter.Level1Extend && (bestBid != null || bestAsk != null))
 		{
-			security ??= EnsureGetSecurity(message);
+			security ??= await EnsureGetSecurityAsync(message, cancellationToken);
 
 			var info = _entityCache.GetSecurityValues(security, time);
 
@@ -1183,7 +1183,7 @@ private void ProcessQuotesMessage(QuoteChangeMessage message)
 #pragma warning disable CS0618 // Type or member is obsolete
 		if (UpdateSecurityLastQuotes)
 		{
-			security ??= EnsureGetSecurity(message);
+			security ??= await EnsureGetSecurityAsync(message, cancellationToken);
 
 			var updated = false;
 
@@ -1214,11 +1214,11 @@ private void ProcessQuotesMessage(QuoteChangeMessage message)
 						if (bid.BoardCode.IsEmpty())
 							continue;
 
-						var innerSecurity = GetSecurity(new SecurityId
+						var innerSecurity = await GetSecurityAsync(new SecurityId
 						{
 							SecurityCode = security.Code,
 							BoardCode = bid.BoardCode
-						});
+						}, cancellationToken);
 
 						var info = changedSecurities.SafeAdd(innerSecurity);
 
@@ -1237,11 +1237,11 @@ private void ProcessQuotesMessage(QuoteChangeMessage message)
 						if (ask.BoardCode.IsEmpty())
 							continue;
 
-						var innerSecurity = GetSecurity(new SecurityId
+						var innerSecurity = await GetSecurityAsync(new SecurityId
 						{
 							SecurityCode = security.Code,
 							BoardCode = ask.BoardCode
-						});
+						}, cancellationToken);
 
 						var info = changedSecurities.SafeAdd(innerSecurity);
 
@@ -1266,7 +1266,7 @@ private void ProcessOrderLogMessage(ExecutionMessage message)
 			return;
 	}
 
-	private void ProcessTradeMessage(ExecutionMessage message)
+	private async ValueTask ProcessTradeMessage(ExecutionMessage message, CancellationToken cancellationToken)
 	{
 		if (RaiseReceived(message, message, TickTradeReceived) != true)
 			return;
@@ -1275,7 +1275,7 @@ private void ProcessTradeMessage(ExecutionMessage message)
 
 		if (ValuesChanged is not null)
 		{
-			security ??= EnsureGetSecurity(message);
+			security ??= await EnsureGetSecurityAsync(message, cancellationToken);
 
 			var time = message.ServerTime;
 			var info = _entityCache.GetSecurityValues(security, time);
@@ -1335,7 +1335,7 @@ private void ProcessTradeMessage(ExecutionMessage message)
 #pragma warning disable CS0618 // Type or member is obsolete
 		if (UpdateSecurityLastQuotes)
 		{
-			security ??= EnsureGetSecurity(message);
+			security ??= await EnsureGetSecurityAsync(message, cancellationToken);
 
 			security.LastTick = message;
 		}
@@ -1438,7 +1438,7 @@ private void ProcessOwnTradeMessage(Order order, Security security, ExecutionMes
 		RaiseReceived(trade, message, OwnTradeReceived);
 	}
 
-	private void ProcessTransactionMessage(ExecutionMessage message)
+	private async ValueTask ProcessTransactionMessage(ExecutionMessage message, CancellationToken cancellationToken)
 	{
 		var originId = message.OriginalTransactionId;
 
@@ -1492,7 +1492,7 @@ private void ProcessTransactionMessage(ExecutionMessage message)
 				return;
 			}
 
-			security = EnsureGetSecurity(message);
+			security = await EnsureGetSecurityAsync(message, cancellationToken);
 
 			if (transactionId == 0 && isStatusRequest)
 				transactionId = TransactionIdGenerator.GetNextId();
@@ -1520,12 +1520,12 @@ private void ProcessTransactionMessage(ExecutionMessage message)
 			throw new ArgumentOutOfRangeException(nameof(message), message.DataType, LocalizedStrings.UnknownType.Put(message));
 	}
 
-	private void ProcessExecutionMessage(ExecutionMessage message)
+	private async ValueTask ProcessExecutionMessage(ExecutionMessage message, CancellationToken cancellationToken)
 	{
 		if (message.DataType == DataType.Transactions)
-			ProcessTransactionMessage(message);
+			await ProcessTransactionMessage(message, cancellationToken);
 		else if (message.DataType == DataType.Ticks)
-			ProcessTradeMessage(message);
+			await ProcessTradeMessage(message, cancellationToken);
 		else if (message.DataType == DataType.OrderLog)
 			ProcessOrderLogMessage(message);
 		else

Algo/Storages/StorageHelper_Obsolete.cs

@@ -215,7 +215,7 @@ public static IMarketDataStorage<CandleMessage> GetCandleMessageStorage(this ISt
 	/// <param name="security">Security.</param>
 	/// <param name="forced">Forced update.</param>
 	/// <remarks>Calls async method via <see cref="AsyncHelper.Run{T}(Func{ValueTask{T}})"/> for backward compatibility.</remarks>
-	//[Obsolete("Use SaveAsync method instead.")]
+	[Obsolete("Use SaveAsync method instead.")]
 	public static void Save(this ISecurityStorage storage, Security security, bool forced)
 	{
 		if (storage is null)

Algo/Storages/StorageMetaInfoMessageAdapter.cs

@@ -90,7 +90,7 @@ protected override async ValueTask OnInnerAdapterNewOutMessageAsync(Message mess
 				else
 					security.ApplyChanges(secMsg, _exchangeInfoProvider, OverrideSecurityData);
 
-				_securityStorage.Save(security, OverrideSecurityData);
+				await _securityStorage.SaveAsync(security, OverrideSecurityData, cancellationToken);
 				break;
 			}
 			case MessageTypes.Board:
@@ -153,7 +153,7 @@ protected override async ValueTask OnInnerAdapterNewOutMessageAsync(Message mess
 				}
 				else
 				{
-					var position = GetPosition(positionMsg.SecurityId, positionMsg.PortfolioName, positionMsg.StrategyId, positionMsg.Side);
+					var position = await GetPositionAsync(positionMsg.SecurityId, positionMsg.PortfolioName, positionMsg.StrategyId, positionMsg.Side, cancellationToken);
 
 					if (position == null)
 						break;
@@ -264,12 +264,12 @@ async ValueTask SendOutAsync<TMessage>(TMessage outMsg)
 		await base.OnSendInMessageAsync(msg, cancellationToken);
 	}
 
-	private Position GetPosition(SecurityId securityId, string portfolioName, string strategyId, Sides? side)
+	private async ValueTask<Position> GetPositionAsync(SecurityId securityId, string portfolioName, string strategyId, Sides? side, CancellationToken cancellationToken)
 	{
 		var security = (!securityId.SecurityCode.IsEmpty() && !securityId.BoardCode.IsEmpty() ? _securityStorage.LookupById(securityId) : _securityStorage.Lookup(new Security
 		{
 			Code = securityId.SecurityCode,
-		}).FirstOrDefault()) ?? TryCreateSecurity(securityId);
+		}).FirstOrDefault()) ?? await TryCreateSecurityAsync(securityId, cancellationToken);
 
 		if (security == null)
 			return null;
@@ -295,7 +295,7 @@ private Position GetPosition(SecurityId securityId, string portfolioName, string
 		};
 	}
 
-	private Security TryCreateSecurity(SecurityId securityId)
+	private async ValueTask<Security> TryCreateSecurityAsync(SecurityId securityId, CancellationToken cancellationToken)
 	{
 		if (securityId.SecurityCode.IsEmpty() || securityId.BoardCode.IsEmpty())
 			return null;
@@ -307,7 +307,7 @@ private Security TryCreateSecurity(SecurityId securityId)
 			Board = _exchangeInfoProvider.GetOrCreateBoard(securityId.BoardCode),
 		};
 
-		_securityStorage.Save(security, false);
+		await _securityStorage.SaveAsync(security, false, cancellationToken);
 
 		return security;
 	}