Correctly mark our `TorTcpConnection` as closed
MODIFY
WalletWasabi/Tor/Socks5/Pool/TorHttpPool.cs
WalletWasabi/Tor/Socks5/Pool/TorHttpPool.cs
@@ -10,11 +10,10 @@ using System.Text;
using System.Threading;
using System.Threading.Channels;
using System.Threading.Tasks;
-using Microsoft.Extensions.Hosting;
using WalletWasabi.Crypto.Randomness;
using WalletWasabi.Helpers;
using WalletWasabi.Logging;
-using WalletWasabi.Tor.Control.Messages.CircuitStatus;
+using WalletWasabi.Tor.Control.Messages.StreamStatus;
using WalletWasabi.Tor.Http;
using WalletWasabi.Tor.Http.Extensions;
using WalletWasabi.Tor.Http.Models;
@@ -48,6 +47,13 @@ public record TorPrebuildCircuitRequest
public TimeSpan RandomDelay { get; }
}
+/// <summary>Latest Tor stream state update.</summary>
+/// <remarks>
+/// Informs that a Tor stream (corresponds to our <see cref="TorTcpConnection"/>) is currently using
+/// the <paramref name="CircuitId">Tor circuit</paramref> and has a certain <paramref name="Status"/>.
+/// </remarks>
+public record TorStreamInfo(string CircuitId, StreamStatusFlag Status);
+
/// <summary>
/// The pool represents a set of multiple TCP connections to Tor SOCKS5 endpoint that are
/// stored in <see cref="TorTcpConnection"/>s.
@@ -82,15 +88,15 @@ public class TorHttpPool : IAsyncDisposable
/// <summary>Channel with pre-building requests.</summary>
private Channel<TorPrebuildCircuitRequest> PreBuildingRequestChannel { get; }
- /// <summary>Key is always a circuit username. Value is the circuit being built.</summary>
+ /// <summary>Key is always Tor SOCKS5 username. Value is the corresponding latest Tor stream update received over Tor control protocol.</summary>
/// <remarks>All access to this object must be guarded by <see cref="ConnectionsLock"/>.</remarks>
- private Dictionary<string, ICircuit> CircuitsBeingBuilt { get; } = new();
+ private Dictionary<string, TorStreamInfo?> TorStreamsBeingBuilt { get; } = new();
/// <summary>Key is always a URI host. Value is a list of pool connections that can connect to the URI host.</summary>
/// <remarks>All access to this object must be guarded by <see cref="ConnectionsLock"/>.</remarks>
private Dictionary<string, List<TorTcpConnection>> ConnectionPerHost { get; } = new();
- /// <remarks>Lock object to guard <see cref="ConnectionPerHost"/> and <see cref="CircuitsBeingBuilt"/>.</remarks>
+ /// <remarks>Lock object to guard <see cref="ConnectionPerHost"/> and <see cref="TorStreamsBeingBuilt"/>.</remarks>
private object ConnectionsLock { get; } = new();
private TorTcpConnectionFactory TcpConnectionFactory { get; }
@@ -320,7 +326,7 @@ public class TorHttpPool : IAsyncDisposable
{
lock (ConnectionsLock)
{
- CircuitsBeingBuilt[circuit.Name] = circuit;
+ TorStreamsBeingBuilt[circuit.Name] = null;
}
TorTcpConnection? connection = null;
@@ -349,7 +355,13 @@ public class TorHttpPool : IAsyncDisposable
{
lock (ConnectionsLock)
{
- CircuitsBeingBuilt.Remove(circuit.Name);
+ if (TorStreamsBeingBuilt.Remove(circuit.Name, out TorStreamInfo? streamInfo))
+ {
+ if (connection is not null && streamInfo is not null)
+ {
+ connection.SetLatestStreamInfo(streamInfo, ifEmpty: true);
+ }
+ }
if (connection is not null)
{
@@ -371,7 +383,7 @@ public class TorHttpPool : IAsyncDisposable
/// <exception cref="TorConnectionWriteException">When a failure during sending our HTTP(s) request to Tor SOCKS5 occurs.</exception>
/// <exception cref="TorConnectionReadException">When a failure during receiving HTTP response from Tor SOCKS5 occurs.</exception>
- internal virtual async Task<HttpResponseMessage> SendCoreAsync(TorTcpConnection connection, HttpRequestMessage request, CancellationToken token = default)
+ internal virtual async Task<HttpResponseMessage> SendCoreAsync(TorTcpConnection connection, HttpRequestMessage request, CancellationToken token)
{
// https://tools.ietf.org/html/rfc7230#section-2.6
// Intermediaries that process HTTP messages (i.e., all intermediaries
@@ -409,42 +421,43 @@ public class TorHttpPool : IAsyncDisposable
/// Allows to report that a Tor circuit status changed.
/// </summary>
/// <param name="circuitID">Tor circuit ID for logging purposes. Example is: <c>35</c>.</param>
- /// <param name="circuitName">Name of the Tor circuit. Example is: <c>IK1DG1HZCZFEQUTF86O86</c>.</param>
+ /// <param name="streamUsername">Username of the Tor stream. Example is: <c>IK1DG1HZCZFEQUTF86O86</c>.</param>
/// <remarks>
/// Useful to clean up <see cref="ConnectionPerHost"/> so that we do not exhaust <see cref="MaxConnectionsPerHost"/> limit.
/// <para>If client code forgets to dispose <see cref="PersonCircuit"/>, this should help us to recover eventually.</para>
/// </remarks>
- public void ReportCircuitStatus(CircStatus circStatus, string circuitID, string circuitName)
+ public void ReportCircuitStatus(string streamUsername, StreamStatusFlag streamStatus, string circuitID)
{
lock (ConnectionsLock)
{
- switch (circStatus)
- {
- case CircStatus.BUILT:
- {
- if (CircuitsBeingBuilt.TryGetValue(circuitName, out ICircuit? circuit))
+ // If the key is not present, then are not the ones starting that Tor stream.
+ if (TorStreamsBeingBuilt.ContainsKey(streamUsername))
{
- Logger.LogTrace($"Tor circuit was built: #{circuitID} ('{circuit}').");
+ // We can receive multiple such reports potentially. We don't mind.
+ TorStreamsBeingBuilt[streamUsername] = new TorStreamInfo(circuitID, streamStatus);
}
- break;
- }
+ TorTcpConnection? tcpConnection = null;
- case CircStatus.CLOSED:
- {
+ // Find our TCP connection based on username we provided when connecting to Tor SOCKS5.
foreach ((string host, List<TorTcpConnection> tcpConnections) in ConnectionPerHost)
{
- foreach (TorTcpConnection tcpConnection in tcpConnections)
- {
- if (tcpConnection.Circuit.Name == circuitName)
+ tcpConnection = tcpConnections.FirstOrDefault(connection => connection.Circuit.Name == streamUsername);
+
+ if (tcpConnection is not null)
{
- Logger.LogTrace($"Tor circuit was closed: #{circuitID} ('{tcpConnection.Circuit}').");
- tcpConnection.MarkAsToDispose(force: false);
- }
+ break;
}
}
- break;
+ if (tcpConnection is not null)
+ {
+ tcpConnection.SetLatestStreamInfo(new TorStreamInfo(circuitID, streamStatus));
+
+ if (streamStatus is StreamStatusFlag.CLOSED or StreamStatusFlag.FAILED)
+ {
+ Logger.LogTrace($"Tor circuit was closed: #{circuitID} ('{tcpConnection.Circuit}').");
+ tcpConnection.MarkAsToDispose(force: false);
}
}
}

MODIFY
WalletWasabi/Tor/Socks5/TorTcpConnection.cs
WalletWasabi/Tor/Socks5/TorTcpConnection.cs
@@ -2,7 +2,7 @@ using System.Diagnostics;
using System.IO;
using System.Net.Security;
using System.Net.Sockets;
-using System.Threading;
+using WalletWasabi.Tor.Control.Messages.StreamStatus;
using WalletWasabi.Tor.Socks5.Pool;
using WalletWasabi.Tor.Socks5.Pool.Circuits;
@@ -35,12 +35,19 @@ public class TorTcpConnection : IDisposable
AllowRecycling = allowRecycling;
}
- /// <remarks>Lock object to guard <see cref="State"/> property.</remarks>
- private object StateLock { get; } = new();
+ /// <remarks>Lock object to guard <see cref="State"/> and <see cref="TorStreamInfo"/> properties.</remarks>
+ private object DataLock { get; } = new();
- /// <remarks>All access to this property must be guarded by <see cref="StateLock"/>.</remarks>
+ /// <remarks>All access to this property must be guarded by <see cref="DataLock"/>.</remarks>
private TcpConnectionState State { get; set; }
+ /// <summary>Circuit ID string as assigned by Tor for the corresponding Tor stream and Tor stream status information.</summary>
+ /// <remarks>
+ /// Corresponds to <see cref="StreamInfo.CircuitID"/>.
+ /// <para>All access to this property must be guarded by <see cref="DataLock"/>.</para>
+ /// </remarks>
+ private TorStreamInfo? TorStreamInfo { get; set; }
+
/// <summary>Gets whether this pool item can be potentially re-used.</summary>
private bool AllowRecycling { get; }
@@ -50,7 +57,7 @@ public class TorTcpConnection : IDisposable
{
get
{
- lock (StateLock)
+ lock (DataLock)
{
return State == TcpConnectionState.ToDispose || !Circuit.IsActive;
}
@@ -79,7 +86,7 @@ public class TorTcpConnection : IDisposable
/// <summary>Reserve the pool item for an HTTP(s) request so no other consumer can use this pool item.</summary>
public bool TryReserve()
{
- lock (StateLock)
+ lock (DataLock)
{
if (State == TcpConnectionState.FreeToUse)
{
@@ -98,7 +105,7 @@ public class TorTcpConnection : IDisposable
/// <returns>Pool item state after unreserve operation.</returns>
public TcpConnectionState Unreserve()
{
- lock (StateLock)
+ lock (DataLock)
{
Debug.Assert(State == TcpConnectionState.InUse, $"Unexpected state: '{State}'.");
State = AllowRecycling ? TcpConnectionState.FreeToUse : TcpConnectionState.ToDispose;
@@ -106,6 +113,25 @@ public class TorTcpConnection : IDisposable
}
}
+ /// <summary>
+ /// Update the latest Tor stream update.
+ /// </summary>
+ /// <param name="ifEmpty"><c>true</c> if we want to process the update only if no previous update has been processed.</param>
+ /// <remarks>
+ /// <paramref name="ifEmpty"/> is helpful to avoid race when <see cref="TorTcpConnection"/> is being created and we
+ /// don't want to miss an update from Tor control for the <see cref="TorTcpConnection"/>.
+ /// </remarks>
+ public void SetLatestStreamInfo(TorStreamInfo streamInfo, bool ifEmpty = false)
+ {
+ lock (DataLock)
+ {
+ if (!ifEmpty || (ifEmpty && TorStreamInfo is null))
+ {
+ TorStreamInfo = streamInfo;
+ }
+ }
+ }
+
/// <param name="force">
/// <c>true</c> if we know that we have just finished using the TCP connection.
/// <c>false</c> if we got an external piece of information that the TCP connection should be closed.
@@ -113,7 +139,7 @@ public class TorTcpConnection : IDisposable
/// <remarks>Connection transporting an HTTP request that was canceled or otherwise failed must be torn down.</remarks>
public void MarkAsToDispose(bool force = true)
{
- lock (StateLock)
+ lock (DataLock)
{
if (force)
{
@@ -129,9 +155,21 @@ public class TorTcpConnection : IDisposable
/// <inheritdoc/>
public override string ToString()
+ {
+ lock (DataLock)
+ {
+ if (TorStreamInfo is not null)
+ {
+ return (TorStreamInfo.Status == StreamStatusFlag.SUCCEEDED)
+ ? $"{Name}#C{TorStreamInfo.CircuitId}"
+ : $"{Name}#C{TorStreamInfo.CircuitId}#{TorStreamInfo.Status}";
+ }
+ else
{
return Name;
}
+ }
+ }
protected virtual void Dispose(bool disposing)
{

MODIFY
WalletWasabi/Tor/TorMonitor.cs
WalletWasabi/Tor/TorMonitor.cs
@@ -11,6 +11,7 @@ using WalletWasabi.Tor.Control.Messages.CircuitStatus;
using WalletWasabi.Tor.Control.Messages.Events;
using WalletWasabi.Tor.Control.Messages.Events.OrEvents;
using WalletWasabi.Tor.Control.Messages.Events.StatusEvents;
+using WalletWasabi.Tor.Control.Messages.StreamStatus;
using WalletWasabi.Tor.Control.Utils;
using WalletWasabi.Tor.Http;
using WalletWasabi.Tor.Socks5.Exceptions;
@@ -158,10 +159,14 @@ public class TorMonitor : PeriodicRunner
Logger.LogInfo("Tor circuit was established.");
circuitEstablished = true;
}
+ }
+ else if (asyncEvent is StreamEvent streamEvent)
+ {
+ StreamInfo info = streamEvent.StreamInfo;
- if ((info.CircStatus is CircStatus.BUILT or CircStatus.CLOSED) && info.UserName is not null)
+ if (info.UserName is not null)
{
- TorHttpPool.ReportCircuitStatus(info.CircStatus, info.CircuitID, info.UserName);
+ TorHttpPool.ReportCircuitStatus(streamUsername: info.UserName, streamStatus: info.StreamStatus, circuitID: info.CircuitID);
}
}
}

Correctly mark our `TorTcpConnection` as closed