AlphaX/qortal
1
0
Fork 0
forked from Qortal/qortal
Code Issues Pull Requests Projects Releases Wiki Activity

Synchronization improvements (again!)

Browse Source 
Bumped version

Controller no longer uses block height to determine whether to sync
but now uses peer's latest block's timestamp and signature.

Also BlockGenerator checks whether it's generating in isolation using
the same peer info (latest block timestamp and signature).

Added API call POST /admin/forcesync peer-address to help get wayward
nodes back on track.

Unified code around, and calling, Transaction.importAsUnconfirmed().

Tidied code around somelock.tryLock() to be more readable.

Controller (post-sync) now broadcasts new chaintip info if
our latest block's signature has changed, not simply the height.

Network.broadcast() only sends out via outbound peer if node has
more than one connection from the same peer. So Controller would
only update one of the peer records with chaintip info.
Controller now updates all connected peers with the ID when it
receives a HEIGHT or HEIGHT_V2 message.

Added node1 thru node7.mcfamily.io to default peers in Network.

Network ignores first "listen port" entry when receiving peers
list from an outbound-connection peer as it already knows
by virtue of having connected to it!

More network message debug logging (hopefully never to be seen).

[some old code left in, but commented out, for a while]
This commit is contained in:
catbref 2019-06-06 14:30:52 +01:00
parent d910cce807
commit c2e8392f05
11 changed files with 525 additions and 399 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

54
src/main/java/org/qora/api/resource/AdminResource.java
View File

@ -11,6 +11,8 @@ import io.swagger.v3.oas.annotations.responses.ApiResponse;
import io.swagger.v3.oas.annotations.tags.Tag;
import java.io.IOException;
import java.net.InetSocketAddress;
import java.net.UnknownHostException;
import java.nio.file.Files;
import java.nio.file.Paths;
import java.time.LocalDate;
@ -43,11 +45,16 @@ import org.qora.api.model.ActivitySummary;
import org.qora.api.model.NodeInfo;
import org.qora.block.BlockChain;
import org.qora.controller.Controller;
import org.qora.controller.Synchronizer;
import org.qora.controller.Synchronizer.SynchronizationResult;
import org.qora.repository.DataException;
import org.qora.repository.Repository;
import org.qora.repository.RepositoryManager;
import org.qora.data.account.ForgingAccountData;
import org.qora.data.account.ProxyForgerData;
import org.qora.network.Network;
import org.qora.network.Peer;
import org.qora.network.PeerAddress;
import org.qora.utils.Base58;
import com.google.common.collect.Lists;
@ -401,4 +408,51 @@ public class AdminResource {
}
}
@POST
@Path("/forcesync")
@Operation(
summary = "Forcibly synchronize to given peer.",
requestBody = @RequestBody(
required = true,
content = @Content(
mediaType = MediaType.TEXT_PLAIN,
schema = @Schema(
type = "string", example = "node7.mcfamily.io"
)
)
),
responses = {
@ApiResponse(
description = "\"true\"",
content = @Content(mediaType = MediaType.TEXT_PLAIN, schema = @Schema(type = "string"))
)
}
)
@ApiErrors({ApiError.INVALID_DATA, ApiError.REPOSITORY_ISSUE})
public String forceSync(String targetPeerAddress) {
Security.checkApiCallAllowed(request);
try {
// Try to resolve passed address to make things easier
PeerAddress peerAddress = PeerAddress.fromString(targetPeerAddress);
InetSocketAddress resolvedAddress = peerAddress.toSocketAddress();
List<Peer> peers = Network.getInstance().getHandshakedPeers();
Peer targetPeer = peers.stream().filter(peer -> peer.getResolvedAddress().equals(resolvedAddress)).findFirst().orElse(null);
if (targetPeer == null)
throw ApiExceptionFactory.INSTANCE.createException(request, ApiError.INVALID_DATA);
SynchronizationResult syncResult = Synchronizer.getInstance().synchronize(targetPeer, true);
return syncResult.name();
} catch (IllegalArgumentException e) {
throw ApiExceptionFactory.INSTANCE.createException(request, ApiError.INVALID_DATA);
} catch (ApiException e) {
throw e;
} catch (UnknownHostException e) {
throw ApiExceptionFactory.INSTANCE.createException(request, ApiError.INVALID_DATA);
}
}
}

4
src/main/java/org/qora/api/resource/TransactionsResource.java
View File

@ -427,12 +427,10 @@ public class TransactionsResource {
if (!transaction.isSignatureValid())
throw ApiExceptionFactory.INSTANCE.createException(request, ApiError.INVALID_SIGNATURE);
ValidationResult result = transaction.isValidUnconfirmed();
ValidationResult result = transaction.importAsUnconfirmed();
if (result != ValidationResult.OK)
throw createTransactionInvalidException(request, result);
transaction.importAsUnconfirmed();
// Notify controller of new transaction
Controller.getInstance().onNewTransaction(transactionData);

6
src/main/java/org/qora/block/BlockChain.java
View File

@ -386,7 +386,9 @@ public class BlockChain {
public static boolean orphan(int targetHeight) throws DataException {
ReentrantLock blockchainLock = Controller.getInstance().getBlockchainLock();
if (blockchainLock.tryLock())
if (!blockchainLock.tryLock())
return false;
try {
try (final Repository repository = RepositoryManager.getRepository()) {
for (int height = repository.getBlockRepository().getBlockchainHeight(); height > targetHeight; --height) {
@ -403,8 +405,6 @@ public class BlockChain {
} finally {
blockchainLock.unlock();
}
return false;
}
}

22
src/main/java/org/qora/block/BlockGenerator.java
View File

@ -10,9 +10,11 @@ import java.util.stream.Collectors;
import org.apache.logging.log4j.LogManager;
import org.apache.logging.log4j.Logger;
import org.qora.account.PrivateKeyAccount;
import org.qora.account.PublicKeyAccount;
import org.qora.block.Block.ValidationResult;
import org.qora.controller.Controller;
import org.qora.data.account.ForgingAccountData;
import org.qora.data.account.ProxyForgerData;
import org.qora.data.block.BlockData;
import org.qora.data.transaction.TransactionData;
import org.qora.network.Network;
@ -112,7 +114,9 @@ public class BlockGenerator extends Thread {
// Make sure we're the only thread modifying the blockchain
ReentrantLock blockchainLock = Controller.getInstance().getBlockchainLock();
if (blockchainLock.tryLock()) {
if (!blockchainLock.tryLock())
continue;
boolean newBlockGenerated = false;
generation: try {
@ -162,9 +166,24 @@ public class BlockGenerator extends Thread {
// Add to blockchain - something else will notice and broadcast new block to network
try {
newBlock.process();
LOGGER.info("Generated new block: " + newBlock.getBlockData().getHeight());
repository.saveChanges();
ProxyForgerData proxyForgerData = repository.getAccountRepository().getProxyForgeData(newBlock.getBlockData().getGeneratorPublicKey());
if (proxyForgerData != null) {
PublicKeyAccount forger = new PublicKeyAccount(repository, proxyForgerData.getForgerPublicKey());
LOGGER.info(String.format("Generated block %d by %s on behalf of %s",
newBlock.getBlockData().getHeight(),
forger.getAddress(),
proxyForgerData.getRecipient()));
} else {
LOGGER.info(String.format("Generated block %d by %s", newBlock.getBlockData().getHeight(), newBlock.getGenerator().getAddress()));
}
repository.saveChanges();
// Notify controller
newBlockGenerated = true;
} catch (DataException e) {
@ -179,7 +198,6 @@ public class BlockGenerator extends Thread {
if (newBlockGenerated)
Controller.getInstance().onGeneratedBlock();
}
}
} catch (DataException e) {
LOGGER.warn("Repository issue while running block generator", e);
}

166
src/main/java/org/qora/controller/Controller.java
View File

@ -69,6 +69,7 @@ public class Controller extends Thread {
private static final Logger LOGGER = LogManager.getLogger(Controller.class);
private static final long MISBEHAVIOUR_COOLOFF = 60 * 60 * 1000; // ms
private static final int MAX_BLOCKCHAIN_TIP_AGE = 5; // blocks
private static final Object shutdownLock = new Object();
private static final String repositoryUrlTemplate = "jdbc:hsqldb:file:%s/blockchain;create=true";
@ -272,23 +273,27 @@ public class Controller extends Thread {
if (peers.size() < Settings.getInstance().getMinBlockchainPeers())
return;
for(Peer peer : peers)
LOGGER.trace(String.format("Peer %s is at height %d", peer, peer.getPeerData().getLastHeight()));
// Remove peers with unknown height, lower height or same height and same block signature (unless we don't have their block signature)
peers.removeIf(hasShorterBlockchain());
// Remove peers that have "misbehaved" recently
// Disregard peers that have "misbehaved" recently
peers.removeIf(hasPeerMisbehaved);
if (!peers.isEmpty()) {
int ourHeight = getChainHeight();
// Remove peers with unknown height, lower height or same height and same block signature (unless we don't have their block signature)
// peers.removeIf(hasShorterBlockchain());
// Disregard peers that don't have a recent block
final long minLatestBlockTimestamp = getMinimumLatestBlockTimestamp();
peers.removeIf(peer -> peer.getPeerData().getLastBlockTimestamp() == null || peer.getPeerData().getLastBlockTimestamp() < minLatestBlockTimestamp);
BlockData latestBlockData = getChainTip();
// Disregard peers that have no block signature or the same block signature as us
peers.removeIf(peer -> peer.getPeerData().getLastBlockSignature() == null || Arrays.equals(latestBlockData.getSignature(), peer.getPeerData().getLastBlockSignature()));
if (!peers.isEmpty()) {
// Pick random peer to sync with
int index = new SecureRandom().nextInt(peers.size());
Peer peer = peers.get(index);
SynchronizationResult syncResult = Synchronizer.getInstance().synchronize(peer);
SynchronizationResult syncResult = Synchronizer.getInstance().synchronize(peer, false);
switch (syncResult) {
case GENESIS_ONLY:
case NO_COMMON_BLOCK:
@ -326,10 +331,10 @@ public class Controller extends Thread {
break;
}
// Broadcast our new height (if changed)
BlockData latestBlockData = getChainTip();
if (latestBlockData.getHeight() != ourHeight)
Network.getInstance().broadcast(recipientPeer -> Network.getInstance().buildHeightMessage(recipientPeer, latestBlockData));
// Broadcast our new chain tip (if changed)
BlockData newLatestBlockData = getChainTip();
if (!Arrays.equals(newLatestBlockData.getSignature(), latestBlockData.getSignature()))
Network.getInstance().broadcast(recipientPeer -> Network.getInstance().buildHeightMessage(recipientPeer, newLatestBlockData));
}
}
@ -457,17 +462,24 @@ public class Controller extends Thread {
case HEIGHT: {
HeightMessage heightMessage = (HeightMessage) message;
// Update our record of peer's height
PeerData peerData = peer.getPeerData();
peer.getPeerData().setLastHeight(heightMessage.getHeight());
// Update all peers with same ID
List<Peer> connectedPeers = Network.getInstance().getHandshakedPeers();
for (Peer connectedPeer : connectedPeers) {
if (connectedPeer.getPeerId() == null || !Arrays.equals(connectedPeer.getPeerId(), peer.getPeerId()))
continue;
PeerData peerData = connectedPeer.getPeerData();
peerData.setLastHeight(heightMessage.getHeight());
// Only save to repository if outbound peer
if (peer.isOutbound())
if (connectedPeer.isOutbound())
try (final Repository repository = RepositoryManager.getRepository()) {
repository.getNetworkRepository().save(peerData);
repository.saveChanges();
} catch (DataException e) {
LOGGER.error(String.format("Repository issue while updating height of peer %s", peer), e);
LOGGER.error(String.format("Repository issue while updating height of peer %s", connectedPeer), e);
}
}
// Potentially synchronize
@ -479,11 +491,17 @@ public class Controller extends Thread {
case HEIGHT_V2: {
HeightV2Message heightV2Message = (HeightV2Message) message;
// Update our record for peer's blockchain info
PeerData peerData = peer.getPeerData();
// Update all peers with same ID
List<Peer> connectedPeers = Network.getInstance().getHandshakedPeers();
for (Peer connectedPeer : connectedPeers) {
if (connectedPeer.getPeerId() == null || !Arrays.equals(connectedPeer.getPeerId(), peer.getPeerId()))
continue;
PeerData peerData = connectedPeer.getPeerData();
// We want to update atomically so use lock
ReentrantLock peerDataLock = peer.getPeerDataLock();
ReentrantLock peerDataLock = connectedPeer.getPeerDataLock();
peerDataLock.lock();
try {
peerData.setLastHeight(heightV2Message.getHeight());
@ -495,12 +513,13 @@ public class Controller extends Thread {
}
// Only save to repository if outbound peer
if (peer.isOutbound())
if (connectedPeer.isOutbound())
try (final Repository repository = RepositoryManager.getRepository()) {
repository.getNetworkRepository().save(peerData);
repository.saveChanges();
} catch (DataException e) {
LOGGER.error(String.format("Repository issue while updating info of peer %s", peer), e);
LOGGER.error(String.format("Repository issue while updating info of peer %s", connectedPeer), e);
}
}
// Potentially synchronize
@ -622,34 +641,28 @@ public class Controller extends Thread {
// Check signature
if (!transaction.isSignatureValid()) {
LOGGER.trace(String.format("Ignoring TRANSACTION %s with invalid signature from peer %s", Base58.encode(transactionData.getSignature()), peer));
LOGGER.trace(String.format("Ignoring %s transaction %s with invalid signature from peer %s", transactionData.getType().name(), Base58.encode(transactionData.getSignature()), peer));
break;
}
// Blockchain lock required to prevent multiple threads trying to save the same transaction simultaneously
ReentrantLock blockchainLock = Controller.getInstance().getBlockchainLock();
if (blockchainLock.tryLock())
try {
// Do we have it already?
if (repository.getTransactionRepository().exists(transactionData.getSignature())) {
LOGGER.trace(String.format("Ignoring existing TRANSACTION %s from peer %s", Base58.encode(transactionData.getSignature()), peer));
ValidationResult validationResult = transaction.importAsUnconfirmed();
if (validationResult == ValidationResult.TRANSACTION_ALREADY_EXISTS) {
LOGGER.trace(String.format("Ignoring existing transaction %s from peer %s", Base58.encode(transactionData.getSignature()), peer));
break;
}
if (validationResult == ValidationResult.NO_BLOCKCHAIN_LOCK) {
LOGGER.trace(String.format("Couldn't lock blockchain to import unconfirmed transaction %s from peer %s", Base58.encode(transactionData.getSignature()), peer));
break;
}
// Is it valid?
ValidationResult validationResult = transaction.isValidUnconfirmed();
if (validationResult != ValidationResult.OK) {
LOGGER.trace(String.format("Ignoring invalid (%s) TRANSACTION %s from peer %s", validationResult.name(), Base58.encode(transactionData.getSignature()), peer));
LOGGER.trace(String.format("Ignoring invalid (%s) %s transaction %s from peer %s", validationResult.name(), transactionData.getType().name(), Base58.encode(transactionData.getSignature()), peer));
break;
}
// Seems ok - add to unconfirmed pile
transaction.importAsUnconfirmed();
LOGGER.debug(String.format("Imported %s transaction %s from peer %s", transactionData.getType().name(), Base58.encode(transactionData.getSignature()), peer));
} finally {
blockchainLock.unlock();
}
} catch (DataException e) {
LOGGER.error(String.format("Repository issue while processing transaction %s from peer %s", Base58.encode(transactionData.getSignature()), peer), e);
}
@ -677,9 +690,9 @@ public class Controller extends Thread {
try (final Repository repository = RepositoryManager.getRepository()) {
for (byte[] signature : signatures) {
// Do we have it already?
// Do we have it already? (Before requesting transaction data itself)
if (repository.getTransactionRepository().exists(signature)) {
LOGGER.trace(String.format("Ignoring unconfirmed transaction %s from peer %s", Base58.encode(signature), peer));
LOGGER.trace(String.format("Ignoring existing transaction %s from peer %s", Base58.encode(signature), peer));
break;
}
@ -697,40 +710,31 @@ public class Controller extends Thread {
// Check signature
if (!transaction.isSignatureValid()) {
LOGGER.trace(String.format("Ignoring unconfirmed transaction %s with invalid signature from peer %s", Base58.encode(transactionData.getSignature()), peer));
LOGGER.trace(String.format("Ignoring %s transaction %s with invalid signature from peer %s", transactionData.getType().name(), Base58.encode(transactionData.getSignature()), peer));
break;
}
// Blockchain lock required to prevent multiple threads trying to save the same transaction simultaneously
ReentrantLock blockchainLock = Controller.getInstance().getBlockchainLock();
if (blockchainLock.tryLock())
try {
// Do we have it already? Rechecking in case it has appeared since previous check above
if (repository.getTransactionRepository().exists(transactionData.getSignature())) {
LOGGER.trace(String.format("Ignoring existing unconfirmed transaction %s from peer %s", Base58.encode(transactionData.getSignature()), peer));
ValidationResult validationResult = transaction.importAsUnconfirmed();
if (validationResult == ValidationResult.TRANSACTION_ALREADY_EXISTS) {
LOGGER.trace(String.format("Ignoring existing transaction %s from peer %s", Base58.encode(transactionData.getSignature()), peer));
break;
}
if (validationResult == ValidationResult.NO_BLOCKCHAIN_LOCK) {
LOGGER.trace(String.format("Couldn't lock blockchain to import unconfirmed transaction %s from peer %s", Base58.encode(transactionData.getSignature()), peer));
break;
}
// Is it valid?
ValidationResult validationResult = transaction.isValidUnconfirmed();
if (validationResult != ValidationResult.OK) {
LOGGER.trace(String.format("Ignoring invalid (%s) unconfirmed transaction %s from peer %s", validationResult.name(), Base58.encode(transactionData.getSignature()), peer));
LOGGER.trace(String.format("Ignoring invalid (%s) %s transaction %s from peer %s", validationResult.name(), transactionData.getType().name(), Base58.encode(transactionData.getSignature()), peer));
break;
}
// Clean repository state before import
repository.discardChanges();
// Seems ok - add to unconfirmed pile
transaction.importAsUnconfirmed();
LOGGER.debug(String.format("Imported %s transaction %s from peer %s", transactionData.getType().name(), Base58.encode(transactionData.getSignature()), peer));
// We could collate signatures that are new to us and broadcast them to our peers too
newSignatures.add(signature);
} finally {
blockchainLock.unlock();
}
}
} catch (DataException e) {
LOGGER.error(String.format("Repository issue while processing unconfirmed transactions from peer %s", peer), e);
@ -809,26 +813,42 @@ public class Controller extends Thread {
};
}
/** Returns whether we think our node has up-to-date blockchain based on our height info about other peers. */
/** Returns whether we think our node has up-to-date blockchain based on our info about other peers. */
public boolean isUpToDate() {
// Is our blockchain too old?
final long minLatestBlockTimestamp = getMinimumLatestBlockTimestamp();
BlockData latestBlockData = getChainTip();
if (latestBlockData.getTimestamp() < minLatestBlockTimestamp)
return false;
List<Peer> peers = Network.getInstance().getUniqueHandshakedPeers();
// Check we have enough peers to potentially synchronize/generator
if (peers.size() < Settings.getInstance().getMinBlockchainPeers())
return false;
// Remove peers with unknown height, lower height or same height and same block signature (unless we don't have their block signature)
peers.removeIf(hasShorterBlockchain());
// Remove peers that within 1 block of our height (actually ourHeight + 1)
final int maxHeight = getChainHeight() + 1;
peers.removeIf(peer -> peer.getPeerData().getLastHeight() <= maxHeight );
// Remove peers that have "misbehaved" recently
// Disregard peers that have "misbehaved" recently
peers.removeIf(hasPeerMisbehaved);
// Disregard peers with unknown height, lower height or same height and same block signature (unless we don't have their block signature)
// peers.removeIf(hasShorterBlockchain());
// Disregard peers that within 1 block of our height (actually ourHeight + 1)
// final int maxHeight = getChainHeight() + 1;
// peers.removeIf(peer -> peer.getPeerData().getLastHeight() <= maxHeight );
// Disregard peers that don't have a recent block
peers.removeIf(peer -> peer.getPeerData().getLastBlockTimestamp() == null || peer.getPeerData().getLastBlockTimestamp() < minLatestBlockTimestamp);
// If we have any peers left, then they would be candidates for synchronization therefore we're not up to date.
return peers.isEmpty();
// return peers.isEmpty();
// If we don't have any peers left then can't synchronize, therefore consider ourself not up to date
return !peers.isEmpty();
}
public long getMinimumLatestBlockTimestamp() {
return NTP.getTime() - BlockChain.getInstance().getMaxBlockTime() * 1000L * MAX_BLOCKCHAIN_TIP_AGE;
}
}

16
src/main/java/org/qora/controller/Synchronizer.java
View File

@ -1,6 +1,5 @@
package org.qora.controller;
import java.util.ArrayList;
import java.util.Arrays;
import java.util.Collections;
import java.util.List;
@ -72,11 +71,14 @@ public class Synchronizer {
* @param peer
* @return false if something went wrong, true otherwise.
*/
public SynchronizationResult synchronize(Peer peer) {
public SynchronizationResult synchronize(Peer peer, boolean force) {
// Make sure we're the only thread modifying the blockchain
// If we're already synchronizing with another peer then this will also return fast
ReentrantLock blockchainLock = Controller.getInstance().getBlockchainLock();
if (blockchainLock.tryLock())
if (!blockchainLock.tryLock())
// Wasn't peer's fault we couldn't sync
return SynchronizationResult.NO_BLOCKCHAIN_LOCK;
try {
try (final Repository repository = RepositoryManager.getRepository()) {
try {
@ -93,7 +95,7 @@ public class Synchronizer {
// XXX this may well be obsolete now
// If peer is too far behind us then don't them.
int minHeight = ourHeight - MAXIMUM_HEIGHT_DELTA;
if (peerHeight < minHeight) {
if (!force && peerHeight < minHeight) {
LOGGER.info(String.format("Peer %s height %d is too far behind our height %d", peer, peerHeight, ourHeight));
return SynchronizationResult.TOO_FAR_BEHIND;
}
@ -141,7 +143,7 @@ public class Synchronizer {
// If common block is too far behind us then we're on massively different forks so give up.
int minCommonHeight = ourHeight - MAXIMUM_COMMON_DELTA;
if (commonBlockHeight < minCommonHeight) {
if (!force && commonBlockHeight < minCommonHeight) {
LOGGER.info(String.format("Blockchain too divergent with peer %s", peer));
return SynchronizationResult.TOO_DIVERGENT;
}
@ -149,6 +151,7 @@ public class Synchronizer {
// If we have blocks after common block then decide whether we want to sync (lowest block signature wins)
int highestMutualHeight = Math.min(peerHeight, ourHeight);
// XXX This might be obsolete now
// If our latest block is very old, we're very behind and should ditch our fork.
if (ourInitialHeight > commonBlockHeight && ourLatestBlockData.getTimestamp() < NTP.getTime() - MAXIMUM_TIP_AGE) {
LOGGER.info(String.format("Ditching our chain after height %d as our latest block is very old", commonBlockHeight));
@ -285,9 +288,6 @@ public class Synchronizer {
} finally {
blockchainLock.unlock();
}
// Wasn't peer's fault we couldn't sync
return SynchronizationResult.NO_BLOCKCHAIN_LOCK;
}
/**

15
src/main/java/org/qora/network/Network.java
View File

@ -65,7 +65,15 @@ public class Network extends Thread {
/** Maximum time since last successful connection before a peer is potentially considered "old", in milliseconds. */
private static final long OLD_PEER_CONNECTION_PERIOD = 7 * 24 * 60 * 60 * 1000; // ms
private static final String[] INITIAL_PEERS = new String[] { "node1.qora.org", "node2.qora.org" };
private static final String[] INITIAL_PEERS = new String[] {
"node1.qora.org",
"node2.qora.org",
"node3.qora.org",
"node4.qora.org",
"node5.qora.org",
"node6.qora.org",
"node7.qora.org"
};
public static final int MAX_SIGNATURES_PER_REPLY = 500;
public static final int MAX_BLOCK_SUMMARIES_PER_REPLY = 500;
@ -479,12 +487,15 @@ public class Network extends Thread {
List<PeerAddress> peerV2Addresses = peersV2Message.getPeerAddresses();
// First entry contains remote peer's listen port but empty address.
// Overwrite address with one obtained from socket.
int peerPort = peerV2Addresses.get(0).getPort();
peerV2Addresses.remove(0);
// If inbound peer, use listen port and socket address to recreate first entry
if (!peer.isOutbound()) {
PeerAddress sendingPeerAddress = PeerAddress.fromString(peer.getPeerData().getAddress().getHost() + ":" + peerPort);
LOGGER.trace("PEERS_V2 sending peer's listen address: " + sendingPeerAddress.toString());
peerV2Addresses.add(0, sendingPeerAddress);
}
mergePeers(peerV2Addresses);
break;

4
src/main/java/org/qora/network/Peer.java
View File

@ -275,8 +275,10 @@ public class Peer implements Runnable {
while (true) {
// Wait (up to INACTIVITY_TIMEOUT) for, and parse, incoming message
Message message = Message.fromStream(in);
if (message == null)
if (message == null) {
this.disconnect("null message");
return;
}
LOGGER.trace(String.format("Received %s message with ID %d from peer %s", message.getType().name(), message.getId(), this));

10
src/main/java/org/qora/network/message/Message.java
View File

@ -16,6 +16,7 @@ import java.io.IOException;
import java.lang.reflect.InvocationTargetException;
import java.lang.reflect.Method;
import java.net.SocketTimeoutException;
import java.nio.BufferUnderflowException;
import java.nio.ByteBuffer;
import java.util.Arrays;
@ -101,14 +102,17 @@ public abstract class Message {
return map.get(value);
}
public Message fromBytes(int id, byte[] data) {
public Message fromBytes(int id, byte[] data) throws MessageException {
if (this.fromByteBuffer == null)
return null;
throw new MessageException("Unsupported message type [" + value + "] during conversion from bytes");
try {
return (Message) this.fromByteBuffer.invoke(null, id, data == null ? null : ByteBuffer.wrap(data));
} catch (IllegalAccessException | IllegalArgumentException | InvocationTargetException e) {
return null;
if (e.getCause() instanceof BufferUnderflowException)
throw new MessageException("Byte data too short for " + name() + " message");
throw new MessageException("Internal error with " + name() + " message during conversion from bytes");
}
}
}

23
src/main/java/org/qora/transaction/Transaction.java
View File

@ -233,6 +233,8 @@ public abstract class Transaction {
GROUP_APPROVAL_NOT_REQUIRED(82),
GROUP_APPROVAL_DECIDED(83),
MAXIMUM_PROXY_RELATIONSHIPS(84),
TRANSACTION_ALREADY_EXISTS(85),
NO_BLOCKCHAIN_LOCK(86),
NOT_YET_RELEASED(1000);
public final int value;
@ -827,7 +829,21 @@ public abstract class Transaction {
*
* @throws DataException
*/
public void importAsUnconfirmed() throws DataException {
public ValidationResult importAsUnconfirmed() throws DataException {
// Attempt to acquire blockchain lock
ReentrantLock blockchainLock = Controller.getInstance().getBlockchainLock();
if (!blockchainLock.tryLock())
return ValidationResult.NO_BLOCKCHAIN_LOCK;
try {
// Check transaction doesn't already exist
if (repository.getTransactionRepository().exists(transactionData.getSignature()))
return ValidationResult.TRANSACTION_ALREADY_EXISTS;
ValidationResult validationResult = this.isValidUnconfirmed();
if (validationResult != ValidationResult.OK)
return validationResult;
// Fix up approval status
if (this.needsGroupApproval()) {
transactionData.setApprovalStatus(ApprovalStatus.PENDING);
@ -838,6 +854,11 @@ public abstract class Transaction {
repository.getTransactionRepository().save(transactionData);
repository.getTransactionRepository().unconfirmTransaction(transactionData);
repository.saveChanges();
return ValidationResult.OK;
} finally {
blockchainLock.unlock();
}
}
/**

4
src/test/java/org/qora/test/common/TransactionUtils.java
View File

@ -31,10 +31,8 @@ public class TransactionUtils {
} catch (InterruptedException e) {
}
ValidationResult result = transaction.isValidUnconfirmed();
ValidationResult result = transaction.importAsUnconfirmed();
assertEquals("Transaction invalid", ValidationResult.OK, result);
transaction.importAsUnconfirmed();
}
/** Signs transaction using given account and forges a new block, using "alice" account. */