fix 'trade not found' bug caused by open offer being spent

do not remove open offer with spent funds if reserved for trade
fix concurrent modification exception
This commit is contained in:
woodser 2023-01-14 10:38:37 -05:00
parent 266d129462
commit cd7f176e2b
7 changed files with 72 additions and 60 deletions

View File

@ -225,7 +225,7 @@ class CoreTradesService {
} }
private Optional<Trade> getClosedTrade(String tradeId) { private Optional<Trade> getClosedTrade(String tradeId) {
Optional<Tradable> tradable = closedTradableManager.getTradableById(tradeId); Optional<Tradable> tradable = closedTradableManager.getTradeById(tradeId);
return tradable.filter((t) -> t instanceof Trade).map(value -> (Trade) value); return tradable.filter((t) -> t instanceof Trade).map(value -> (Trade) value);
} }

View File

@ -196,7 +196,7 @@ public class OpenOfferManager implements PeerManager.Listener, DecryptedDirectMe
@Override @Override
public void onAdded(Offer offer) { public void onAdded(Offer offer) {
Optional<OpenOffer> openOfferOptional = getOpenOfferById(offer.getId()); Optional<OpenOffer> openOfferOptional = getOpenOfferById(offer.getId());
if (openOfferOptional.isPresent() && offer.isReservedFundsSpent()) { if (openOfferOptional.isPresent() && openOfferOptional.get().getState() != OpenOffer.State.RESERVED && offer.isReservedFundsSpent()) {
removeOpenOffer(openOfferOptional.get(), null); removeOpenOffer(openOfferOptional.get(), null);
} }
} }
@ -247,8 +247,8 @@ public class OpenOfferManager implements PeerManager.Listener, DecryptedDirectMe
// .ifPresent(errorMsg -> invalidOffers.add(new Tuple2<>(openOffer, errorMsg)))); // .ifPresent(errorMsg -> invalidOffers.add(new Tuple2<>(openOffer, errorMsg))));
// process unposted offers // process unposted offers
processUnpostedOffers((transaction) -> {}, (errMessage) -> { processUnpostedOffers((transaction) -> {}, (errorMessage) -> {
log.warn("Error processing unposted offers on new unlocked balance: " + errMessage); log.warn("Error processing unposted offers on new unlocked balance: " + errorMessage);
}); });
// register to process unposted offers when unlocked balance increases // register to process unposted offers when unlocked balance increases
@ -257,8 +257,8 @@ public class OpenOfferManager implements PeerManager.Listener, DecryptedDirectMe
@Override @Override
public void onBalancesChanged(BigInteger newBalance, BigInteger newUnlockedBalance) { public void onBalancesChanged(BigInteger newBalance, BigInteger newUnlockedBalance) {
if (lastUnlockedBalance == null || lastUnlockedBalance.compareTo(newUnlockedBalance) < 0) { if (lastUnlockedBalance == null || lastUnlockedBalance.compareTo(newUnlockedBalance) < 0) {
processUnpostedOffers((transaction) -> {}, (errMessage) -> { processUnpostedOffers((transaction) -> {}, (errorMessage) -> {
log.warn("Error processing unposted offers on new unlocked balance: " + errMessage); log.warn("Error processing unposted offers on new unlocked balance: " + errorMessage);
}); });
} }
lastUnlockedBalance = newUnlockedBalance; lastUnlockedBalance = newUnlockedBalance;
@ -327,6 +327,7 @@ public class OpenOfferManager implements PeerManager.Listener, DecryptedDirectMe
List<OpenOffer> openOffersList = new ArrayList<>(openOffers); List<OpenOffer> openOffersList = new ArrayList<>(openOffers);
openOffersList.forEach(openOffer -> removeOpenOffer(openOffer, () -> { openOffersList.forEach(openOffer -> removeOpenOffer(openOffer, () -> {
}, errorMessage -> { }, errorMessage -> {
log.warn("Error removing open offer: " + errorMessage);
})); }));
if (completeHandler != null) if (completeHandler != null)
UserThread.runAfter(completeHandler, size * 200 + 500, TimeUnit.MILLISECONDS); UserThread.runAfter(completeHandler, size * 200 + 500, TimeUnit.MILLISECONDS);
@ -448,10 +449,11 @@ public class OpenOfferManager implements PeerManager.Listener, DecryptedDirectMe
addOpenOffer(openOffer); addOpenOffer(openOffer);
requestPersistence(); requestPersistence();
resultHandler.handleResult(transaction); resultHandler.handleResult(transaction);
}, (errMessage) -> { }, (errorMessage) -> {
log.warn("Error processing unposted offer {}: {}", openOffer.getId(), errorMessage);
onRemoved(openOffer); onRemoved(openOffer);
offer.setErrorMessage(errMessage); offer.setErrorMessage(errorMessage);
errorMessageHandler.handleErrorMessage(errMessage); errorMessageHandler.handleErrorMessage(errorMessage);
}); });
} }
@ -691,6 +693,7 @@ public class OpenOfferManager implements PeerManager.Listener, DecryptedDirectMe
processUnpostedOffer(openOffers, scheduledOffer, (transaction) -> { processUnpostedOffer(openOffers, scheduledOffer, (transaction) -> {
latch.countDown(); latch.countDown();
}, errorMessage -> { }, errorMessage -> {
log.warn("Error processing unposted offer {}: {}", scheduledOffer.getId(), errorMessage);
onRemoved(scheduledOffer); onRemoved(scheduledOffer);
errorMessages.add(errorMessage); errorMessages.add(errorMessage);
latch.countDown(); latch.countDown();

View File

@ -151,6 +151,10 @@ public class ClosedTradableManager implements PersistedDataHost {
return closedTradables.stream().filter(e -> e.getId().equals(id)).findFirst(); return closedTradables.stream().filter(e -> e.getId().equals(id)).findFirst();
} }
public Optional<Tradable> getTradeById(String id) {
return closedTradables.stream().filter(e -> e instanceof Trade && e.getId().equals(id)).findFirst();
}
public void maybeClearSensitiveData() { public void maybeClearSensitiveData() {
log.info("checking closed trades eligibility for having sensitive data cleared"); log.info("checking closed trades eligibility for having sensitive data cleared");
closedTradables.stream() closedTradables.stream()

View File

@ -20,7 +20,6 @@ package bisq.core.trade;
import bisq.core.monetary.Price; import bisq.core.monetary.Price;
import bisq.core.monetary.Volume; import bisq.core.monetary.Volume;
import bisq.core.offer.Offer; import bisq.core.offer.Offer;
import bisq.core.trade.protocol.TradingPeer;
import bisq.network.p2p.NodeAddress; import bisq.network.p2p.NodeAddress;
import bisq.common.proto.persistable.PersistablePayload; import bisq.common.proto.persistable.PersistablePayload;

View File

@ -895,11 +895,14 @@ public class TradeManager implements PersistedDataHost, DecryptedDirectMessageLi
} }
public Stream<Trade> getTradesStreamWithFundsLockedIn() { public Stream<Trade> getTradesStreamWithFundsLockedIn() {
synchronized (tradableList) {
return getObservableList().stream().filter(Trade::isFundsLockedIn); return getObservableList().stream().filter(Trade::isFundsLockedIn);
} }
}
public Set<String> getSetOfFailedOrClosedTradeIdsFromLockedInFunds() throws TradeTxException { public Set<String> getSetOfFailedOrClosedTradeIdsFromLockedInFunds() throws TradeTxException {
AtomicReference<TradeTxException> tradeTxException = new AtomicReference<>(); AtomicReference<TradeTxException> tradeTxException = new AtomicReference<>();
synchronized (tradableList) {
Set<String> tradesIdSet = getTradesStreamWithFundsLockedIn() Set<String> tradesIdSet = getTradesStreamWithFundsLockedIn()
.filter(Trade::hasFailed) .filter(Trade::hasFailed)
.map(Trade::getId) .map(Trade::getId)
@ -948,6 +951,7 @@ public class TradeManager implements PersistedDataHost, DecryptedDirectMessageLi
return tradesIdSet; return tradesIdSet;
} }
}
// If trade still has funds locked up it might come back from failed trades // If trade still has funds locked up it might come back from failed trades
// Aborts unfailing if the address entries needed are not available // Aborts unfailing if the address entries needed are not available
@ -1028,11 +1032,13 @@ public class TradeManager implements PersistedDataHost, DecryptedDirectMessageLi
} }
public List<Trade> getOpenTrades() { public List<Trade> getOpenTrades() {
synchronized (tradableList) {
return ImmutableList.copyOf(getObservableList().stream() return ImmutableList.copyOf(getObservableList().stream()
.filter(e -> e instanceof Trade) .filter(e -> e instanceof Trade)
.map(e -> e) .map(e -> e)
.collect(Collectors.toList())); .collect(Collectors.toList()));
} }
}
public Optional<Trade> getClosedTrade(String tradeId) { public Optional<Trade> getClosedTrade(String tradeId) {
return closedTradableManager.getClosedTrades().stream().filter(e -> e.getId().equals(tradeId)).findFirst(); return closedTradableManager.getClosedTrades().stream().filter(e -> e.getId().equals(tradeId)).findFirst();

View File

@ -87,6 +87,7 @@ class GrpcTradesService extends TradesImplBase {
} catch (IllegalArgumentException cause) { } catch (IllegalArgumentException cause) {
// Offer makers may call 'gettrade' many times before a trade exists. // Offer makers may call 'gettrade' many times before a trade exists.
// Log a 'trade not found' warning instead of a full stack trace. // Log a 'trade not found' warning instead of a full stack trace.
cause.printStackTrace();
exceptionHandler.handleExceptionAsWarning(log, "getTrade", cause, responseObserver); exceptionHandler.handleExceptionAsWarning(log, "getTrade", cause, responseObserver);
} catch (Throwable cause) { } catch (Throwable cause) {
exceptionHandler.handleException(log, cause, responseObserver); exceptionHandler.handleException(log, cause, responseObserver);

View File

@ -20,7 +20,6 @@ package bisq.desktop.main.portfolio.closedtrades;
import bisq.desktop.common.model.ActivatableWithDataModel; import bisq.desktop.common.model.ActivatableWithDataModel;
import bisq.desktop.common.model.ViewModel; import bisq.desktop.common.model.ViewModel;
import bisq.core.monetary.Volume;
import bisq.core.trade.ClosedTradableFormatter; import bisq.core.trade.ClosedTradableFormatter;
import org.bitcoinj.core.Coin; import org.bitcoinj.core.Coin;