Implement client side CJ
MODIFY
WalletWasabi.Gui/Global.cs
WalletWasabi.Gui/Global.cs
@@ -21,6 +21,7 @@ using WalletWasabi.Blockchain.Analysis.FeesEstimation;
using WalletWasabi.Blockchain.Keys;
using WalletWasabi.Blockchain.TransactionBroadcasting;
using WalletWasabi.Blockchain.TransactionOutputs;
+using WalletWasabi.CoinJoin.Client;
using WalletWasabi.CoinJoin.Client.Clients;
using WalletWasabi.Gui.Helpers;
using WalletWasabi.Helpers;
@@ -58,6 +59,7 @@ namespace WalletWasabi.Gui
public CoinJoinClient ChaumianClient { get; private set; }
public WalletService WalletService { get; private set; }
public TransactionBroadcaster TransactionBroadcaster { get; set; }
+ public CoinJoinProcessor CoinJoinProcessor { get; set; }
public Node RegTestMempoolServingNode { get; private set; }
public TorProcessManager TorManager { get; private set; }
public CoreNode BitcoinCoreNode { get; private set; }
@@ -346,6 +348,7 @@ namespace WalletWasabi.Gui
#endregion SynchronizerInitialization
TransactionBroadcaster = new TransactionBroadcaster(Network, BitcoinStore, Synchronizer, Nodes, BitcoinCoreNode?.RpcClient);
+ CoinJoinProcessor = new CoinJoinProcessor(Synchronizer, BitcoinCoreNode?.RpcClient);
}
finally
{
@@ -474,6 +477,7 @@ namespace WalletWasabi.Gui
WalletService.TransactionProcessor.CoinReceived += CoinReceived;
TransactionBroadcaster.AddWalletService(WalletService);
+ CoinJoinProcessor.AddWalletService(WalletService);
}
_cancelWalletServiceInitialization = null; // Must make it null explicitly, because dispose won't make it null.
}
@@ -639,6 +643,13 @@ namespace WalletWasabi.Gui
Logger.LogInfo($"Disposed {nameof(FeeProviders)}.");
}
+ var coinJoinProcessor = CoinJoinProcessor;
+ if (coinJoinProcessor is { })
+ {
+ coinJoinProcessor.Dispose();
+ Logger.LogInfo($"{nameof(CoinJoinProcessor)} is disposed.");
+ }
+
var synchronizer = Synchronizer;
if (synchronizer is { })
{

ADD
WalletWasabi/CoinJoin/Client/CoinJoinProcessor.cs
+using NBitcoin;
+using NBitcoin.RPC;
+using Nito.AsyncEx;
+using System;
+using System.Collections.Generic;
+using System.Linq;
+using System.Net.Http;
+using System.Text;
+using System.Threading;
+using System.Threading.Tasks;
+using WalletWasabi.Backend.Models.Responses;
+using WalletWasabi.Blockchain.Transactions;
+using WalletWasabi.Helpers;
+using WalletWasabi.Logging;
+using WalletWasabi.Models;
+using WalletWasabi.Services;
+using WalletWasabi.WebClients.Wasabi;
+
+namespace WalletWasabi.CoinJoin.Client
+{
+ public class CoinJoinProcessor : IDisposable
+ {
+ public WasabiSynchronizer Synchronizer { get; }
+ public Dictionary<WalletService, HashSet<uint256>> WalletServices { get; }
+ public IEnumerable<KeyValuePair<WalletService, HashSet<uint256>>> AliveWalletServices => WalletServices.Where(x => x.Key is { IsDisposed: var isDisposed } && !isDisposed);
+ public object WalletServicesLock { get; }
+ public RPCClient RpcClient { get; private set; }
+ private AsyncLock ProcessLock { get; }
+
+ public CoinJoinProcessor(WasabiSynchronizer synchronizer, RPCClient rpc)
+ {
+ Synchronizer = Guard.NotNull(nameof(synchronizer), synchronizer);
+ WalletServices = new Dictionary<WalletService, HashSet<uint256>>();
+ WalletServicesLock = new object();
+ RpcClient = rpc;
+ ProcessLock = new AsyncLock();
+ Synchronizer.ResponseArrived += Synchronizer_ResponseArrivedAsync;
+ }
+
+ private async void Synchronizer_ResponseArrivedAsync(object sender, SynchronizeResponse response)
+ {
+ try
+ {
+ using (await ProcessLock.LockAsync())
+ {
+ var unconfirmedCoinJoinHashes = response.UnconfirmedCoinJoins;
+ if (!unconfirmedCoinJoinHashes.Any())
+ {
+ return;
+ }
+
+ var txsNotKnownByAWalletService = new HashSet<uint256>();
+ lock (WalletServicesLock)
+ {
+ foreach (var pair in AliveWalletServices)
+ {
+ // If a wallet service doesn't know about the tx, then we add it for processing.
+ foreach (var tx in unconfirmedCoinJoinHashes.Where(x => !pair.Value.Contains(x)))
+ {
+ txsNotKnownByAWalletService.Add(tx);
+ }
+ }
+ }
+
+ using var client = new WasabiClient(Synchronizer.WasabiClient.TorClient.DestinationUriAction, Synchronizer.WasabiClient.TorClient.TorSocks5EndPoint);
+ var unconfirmedCoinJoins = await client.GetTransactionsAsync(Synchronizer.Network, txsNotKnownByAWalletService, CancellationToken.None).ConfigureAwait(false);
+
+ foreach (var tx in unconfirmedCoinJoins)
+ {
+ if (RpcClient is null
+ || await TryBroadcastTransactionWithRpcAsync(tx).ConfigureAwait(false)
+ || (await RpcClient.TestAsync().ConfigureAwait(false) is { })) // If the test throws exception then I believe it, because RPC is down and the backend is the god.
+ {
+ BelieveTransaction(tx);
+ }
+ }
+ }
+ }
+ catch (Exception ex)
+ {
+ Logger.LogError(ex);
+ }
+ }
+
+ public void AddWalletService(WalletService walletService)
+ {
+ Guard.NotNull(nameof(walletService), walletService);
+ lock (WalletServicesLock)
+ {
+ WalletServices.Add(walletService, new HashSet<uint256>());
+ }
+ }
+
+ private void BelieveTransaction(Transaction transaction)
+ {
+ lock (WalletServicesLock)
+ lock (TransactionProcessor.Lock)
+ {
+ foreach (var pair in AliveWalletServices.Where(x => !x.Value.Contains(transaction.GetHash())))
+ {
+ var walletService = pair.Key;
+ pair.Value.Add(transaction.GetHash());
+ walletService.TransactionProcessor.Process(new SmartTransaction(transaction, Height.Mempool));
+ }
+ }
+ }
+
+ private async Task<bool> TryBroadcastTransactionWithRpcAsync(Transaction transaction)
+ {
+ try
+ {
+ await RpcClient.SendRawTransactionAsync(transaction).ConfigureAwait(false);
+ Logger.LogInfo($"Transaction is successfully broadcasted with RPC: {transaction.GetHash()}.");
+
+ return true;
+ }
+ catch
+ {
+ return false;
+ }
+ }
+
+ #region IDisposable Support
+
+ private volatile bool _disposedValue = false; // To detect redundant calls
+
+ protected virtual void Dispose(bool disposing)
+ {
+ if (!_disposedValue)
+ {
+ if (disposing)
+ {
+ Synchronizer.ResponseArrived -= Synchronizer_ResponseArrivedAsync;
+ }
+
+ _disposedValue = true;
+ }
+ }
+
+ // This code added to correctly implement the disposable pattern.
+ public void Dispose()
+ {
+ // Do not change this code. Put cleanup code in Dispose(bool disposing) above.
+ Dispose(true);
+ }
+
+ #endregion IDisposable Support
+ }
+}

MODIFY
WalletWasabi/WebClients/Wasabi/WasabiClient.cs
WalletWasabi/WebClients/Wasabi/WasabiClient.cs
@@ -81,6 +81,47 @@ namespace WalletWasabi.WebClients.Wasabi
return ret;
}
+ public static Dictionary<uint256, Transaction> TransactionCache { get; } = new Dictionary<uint256, Transaction>();
+ public static object TransactionCacheLock { get; } = new object();
+
+ public async Task<IEnumerable<Transaction>> GetTransactionsAsync(Network network, IEnumerable<uint256> txHashes, CancellationToken cancel)
+ {
+ // todo: order
+ // todo: handle max 10tx
+ // todo: TransactionCache
+ //lock (TransactionCacheLock)
+ //{
+ // var notCachedTxHashses = TransactionCache.Where(x => x.Key)
+ //}
+
+ using var response = await TorClient.SendAndRetryAsync(
+ HttpMethod.Get,
+ HttpStatusCode.OK,
+ $"/api/v{Constants.BackendMajorVersion}/btc/blockchain/transaction-hexes?&transactionIds={string.Join("&transactionIds=", txHashes.Select(x => x.ToString()))}",
+ cancel: cancel);
+ if (response.StatusCode != HttpStatusCode.OK)
+ {
+ await response.ThrowRequestExceptionFromContentAsync();
+ }
+
+ using HttpContent content = response.Content;
+ var retString = await content.ReadAsJsonAsync<IEnumerable<string>>();
+ var ret = retString.Select(x => Transaction.Parse(x, network)).ToList();
+
+ lock (TransactionCacheLock)
+ {
+ foreach (var tx in ret)
+ {
+ if (TransactionCache.TryAdd(tx.GetHash(), tx) && TransactionCache.Count >= 1000)
+ {
+ TransactionCache.Remove(TransactionCache.Keys.First());
+ }
+ }
+ }
+
+ return ret;
+ }
+
public async Task<IDictionary<int, FeeEstimationPair>> GetFeesAsync(params int[] confirmationTargets)
{
var confirmationTargetsString = string.Join(",", confirmationTargets);

Implement client side CJ