SHA256
Обновить синхронизацию серверов и экран сохранения ключей
This commit is contained in:
AidarKC
+348
@@ -0,0 +1,348 @@
|
||||
package server.sync;
|
||||
|
||||
import com.fasterxml.jackson.databind.JsonNode;
|
||||
import com.fasterxml.jackson.databind.ObjectMapper;
|
||||
import org.slf4j.Logger;
|
||||
import org.slf4j.LoggerFactory;
|
||||
import server.logic.ws_protocol.Base64Ws;
|
||||
import shine.db.dao.BlocksDAO;
|
||||
import shine.db.dao.SyncServersDAO;
|
||||
import shine.db.entities.BlockEntry;
|
||||
import shine.db.entities.SyncServerEntry;
|
||||
import utils.blockchain.BlockchainNameUtil;
|
||||
|
||||
import java.net.URI;
|
||||
import java.net.http.HttpClient;
|
||||
import java.net.http.WebSocket;
|
||||
import java.nio.ByteBuffer;
|
||||
import java.time.Duration;
|
||||
import java.util.List;
|
||||
import java.util.Locale;
|
||||
import java.util.UUID;
|
||||
import java.util.concurrent.*;
|
||||
import java.util.concurrent.atomic.AtomicLong;
|
||||
|
||||
/**
|
||||
* Фоновая one-shot репликация AddBlock на серверы из локальной таблицы sync_servers.
|
||||
*/
|
||||
public final class AddBlockSyncService {
|
||||
|
||||
private static final Logger log = LoggerFactory.getLogger(AddBlockSyncService.class);
|
||||
private static final ObjectMapper MAPPER = new ObjectMapper();
|
||||
private static final HttpClient HTTP = HttpClient.newBuilder()
|
||||
.connectTimeout(Duration.ofSeconds(6))
|
||||
.build();
|
||||
private static final ExecutorService EXECUTOR = new ThreadPoolExecutor(
|
||||
1,
|
||||
Math.max(2, Runtime.getRuntime().availableProcessors()),
|
||||
60L,
|
||||
TimeUnit.SECONDS,
|
||||
new LinkedBlockingQueue<>(10_000),
|
||||
new ThreadFactory() {
|
||||
private final AtomicLong n = new AtomicLong(1);
|
||||
@Override
|
||||
public Thread newThread(Runnable r) {
|
||||
Thread t = new Thread(r, "sync-addblock-" + n.getAndIncrement());
|
||||
t.setDaemon(true);
|
||||
return t;
|
||||
}
|
||||
},
|
||||
new ThreadPoolExecutor.DiscardPolicy()
|
||||
);
|
||||
|
||||
private final BlocksDAO blocksDAO = BlocksDAO.getInstance();
|
||||
private final SyncServersDAO syncServersDAO = SyncServersDAO.getInstance();
|
||||
|
||||
public void replicateAsync(String blockchainName, int blockNumber) {
|
||||
EXECUTOR.execute(() -> {
|
||||
try {
|
||||
replicate(blockchainName, blockNumber);
|
||||
} catch (Exception e) {
|
||||
log.error("AddBlock sync failed unexpectedly (blockchainName={}, blockNumber={})",
|
||||
blockchainName, blockNumber, e);
|
||||
}
|
||||
});
|
||||
}
|
||||
|
||||
private void replicate(String blockchainName, int blockNumber) throws Exception {
|
||||
String ownerLogin = normalize(BlockchainNameUtil.loginFromBlockchainName(blockchainName));
|
||||
if (ownerLogin == null) {
|
||||
log.warn("AddBlock sync skipped: cannot derive owner login from blockchainName={}", blockchainName);
|
||||
return;
|
||||
}
|
||||
|
||||
List<SyncServerEntry> partners = syncServersDAO.listAll();
|
||||
if (partners.isEmpty()) {
|
||||
return;
|
||||
}
|
||||
|
||||
BlockEntry currentBlock = blocksDAO.getByNumber(blockchainName, blockNumber);
|
||||
if (currentBlock == null || currentBlock.getBlockBytes() == null) {
|
||||
log.warn("AddBlock sync skipped: block not found in DB (blockchainName={}, blockNumber={})",
|
||||
blockchainName, blockNumber);
|
||||
return;
|
||||
}
|
||||
|
||||
for (SyncServerEntry partner : partners) {
|
||||
if (partner == null) continue;
|
||||
String partnerLogin = normalize(partner.getLogin());
|
||||
if (partnerLogin == null) continue;
|
||||
if (partnerLogin.equals(ownerLogin)) {
|
||||
continue;
|
||||
}
|
||||
try {
|
||||
replicateToPartner(partner, blockchainName, blockNumber, currentBlock);
|
||||
} catch (Exception e) {
|
||||
log.warn("AddBlock sync aborted for partner login={} blockchainName={} blockNumber={} reason={}",
|
||||
partnerLogin, blockchainName, blockNumber, e.toString());
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
private void replicateToPartner(SyncServerEntry partner, String blockchainName, int blockNumber, BlockEntry currentBlock) throws Exception {
|
||||
String wsUrl = buildWsUrl(partner.getServerAddress());
|
||||
if (wsUrl == null) {
|
||||
log.warn("AddBlock sync skipped: invalid server_address for partner login={} address={}",
|
||||
partner.getLogin(), partner.getServerAddress());
|
||||
return;
|
||||
}
|
||||
|
||||
AddBlockPushResult firstTry = pushBlock(wsUrl, blockchainName, currentBlock);
|
||||
if (firstTry.ok()) {
|
||||
log.info("AddBlock sync ok: partner={} blockchainName={} blockNumber={}",
|
||||
partner.getLogin(), blockchainName, blockNumber);
|
||||
return;
|
||||
}
|
||||
|
||||
if (!firstTry.needsBackfill()) {
|
||||
log.warn("AddBlock sync failed without backfill: partner={} blockchainName={} blockNumber={} code={}",
|
||||
partner.getLogin(), blockchainName, blockNumber, firstTry.code());
|
||||
return;
|
||||
}
|
||||
|
||||
int remoteLast = firstTry.serverLastGlobalNumber();
|
||||
int fromBlockNumber = remoteLast + 1;
|
||||
if (fromBlockNumber > blockNumber) {
|
||||
log.warn("AddBlock sync inconsistent backfill window: partner={} blockchainName={} remoteLast={} target={}",
|
||||
partner.getLogin(), blockchainName, remoteLast, blockNumber);
|
||||
return;
|
||||
}
|
||||
|
||||
List<BlockEntry> missingBlocks = blocksDAO.listRangeByNumber(blockchainName, fromBlockNumber, blockNumber);
|
||||
if (missingBlocks.isEmpty()) {
|
||||
log.warn("AddBlock sync backfill failed: local range empty partner={} blockchainName={} from={} to={}",
|
||||
partner.getLogin(), blockchainName, fromBlockNumber, blockNumber);
|
||||
return;
|
||||
}
|
||||
|
||||
for (BlockEntry blockEntry : missingBlocks) {
|
||||
AddBlockPushResult backfillResult = pushBlock(wsUrl, blockchainName, blockEntry);
|
||||
if (!backfillResult.ok()) {
|
||||
log.warn("AddBlock sync backfill failed: partner={} blockchainName={} blockNumber={} code={}",
|
||||
partner.getLogin(), blockchainName, blockEntry.getBlockNumber(), backfillResult.code());
|
||||
return;
|
||||
}
|
||||
}
|
||||
|
||||
log.info("AddBlock sync backfill ok: partner={} blockchainName={} from={} to={}",
|
||||
partner.getLogin(), blockchainName, fromBlockNumber, blockNumber);
|
||||
}
|
||||
|
||||
private AddBlockPushResult pushBlock(String wsUrl, String blockchainName, BlockEntry blockEntry) throws Exception {
|
||||
JsonNode response = sendAddBlock(wsUrl, blockchainName, blockEntry);
|
||||
int status = response.path("status").asInt(500);
|
||||
if (status >= 200 && status < 300) {
|
||||
return AddBlockPushResult.success();
|
||||
}
|
||||
|
||||
String code = textOrEmpty(response, "code");
|
||||
if (code.isBlank()) {
|
||||
code = textOrEmpty(response, "error");
|
||||
}
|
||||
JsonNode payload = response.path("payload");
|
||||
int serverLastGlobalNumber = payload.path("serverLastGlobalNumber").asInt(Integer.MIN_VALUE);
|
||||
String serverLastGlobalHash = payload.path("serverLastGlobalHash").asText("");
|
||||
|
||||
return new AddBlockPushResult(false, status, code, serverLastGlobalNumber, serverLastGlobalHash);
|
||||
}
|
||||
|
||||
private JsonNode sendAddBlock(String wsUrl, String blockchainName, BlockEntry blockEntry) throws Exception {
|
||||
CompletableFuture<String> responseFuture = new CompletableFuture<>();
|
||||
CountDownLatch openLatch = new CountDownLatch(1);
|
||||
SyncWsListener listener = new SyncWsListener(responseFuture, openLatch);
|
||||
|
||||
WebSocket webSocket = HTTP.newWebSocketBuilder()
|
||||
.connectTimeout(Duration.ofSeconds(6))
|
||||
.buildAsync(URI.create(wsUrl), listener)
|
||||
.get(8, TimeUnit.SECONDS);
|
||||
|
||||
if (!openLatch.await(8, TimeUnit.SECONDS)) {
|
||||
tryAbort(webSocket);
|
||||
throw new TimeoutException("WS open timeout");
|
||||
}
|
||||
|
||||
String requestId = "sync-" + UUID.randomUUID();
|
||||
String json = buildAddBlockJson(requestId, blockchainName, blockEntry);
|
||||
webSocket.sendText(json, true).get(8, TimeUnit.SECONDS);
|
||||
|
||||
String responseJson = responseFuture.get(12, TimeUnit.SECONDS);
|
||||
tryAbort(webSocket);
|
||||
return MAPPER.readTree(responseJson);
|
||||
}
|
||||
|
||||
private String buildAddBlockJson(String requestId, String blockchainName, BlockEntry blockEntry) throws Exception {
|
||||
String prevHashHex = blockEntry.getBlockNumber() <= 0
|
||||
? ""
|
||||
: toHex(extractPrevHash32(blockEntry.getBlockBytes()));
|
||||
String blockBytesB64 = Base64Ws.encode(blockEntry.getBlockBytes());
|
||||
|
||||
String safeBlockchainName = MAPPER.writeValueAsString(blockchainName);
|
||||
String safePrevHashHex = MAPPER.writeValueAsString(prevHashHex);
|
||||
String safeBlockBytes = MAPPER.writeValueAsString(blockBytesB64);
|
||||
String safeRequestId = MAPPER.writeValueAsString(requestId);
|
||||
|
||||
return """
|
||||
{
|
||||
"op":"AddBlock",
|
||||
"requestId":%s,
|
||||
"payload":{
|
||||
"blockchainName":%s,
|
||||
"blockNumber":%d,
|
||||
"prevBlockHash":%s,
|
||||
"blockBytesB64":%s
|
||||
}
|
||||
}
|
||||
""".formatted(safeRequestId, safeBlockchainName, blockEntry.getBlockNumber(), safePrevHashHex, safeBlockBytes);
|
||||
}
|
||||
|
||||
private static byte[] extractPrevHash32(byte[] blockBytes) {
|
||||
if (blockBytes == null || blockBytes.length < 44) {
|
||||
return new byte[32];
|
||||
}
|
||||
byte[] out = new byte[32];
|
||||
System.arraycopy(blockBytes, 12, out, 0, 32);
|
||||
return out;
|
||||
}
|
||||
|
||||
private static String textOrEmpty(JsonNode node, String field) {
|
||||
return node == null ? "" : node.path(field).asText("");
|
||||
}
|
||||
|
||||
private static String normalize(String value) {
|
||||
if (value == null) return null;
|
||||
String s = value.trim().toLowerCase(Locale.ROOT);
|
||||
return s.isEmpty() ? null : s;
|
||||
}
|
||||
|
||||
private static String buildWsUrl(String serverAddressRaw) {
|
||||
String host = normalizeHostLike(serverAddressRaw);
|
||||
if (host == null) return null;
|
||||
return "wss://" + host + "/ws";
|
||||
}
|
||||
|
||||
private static String normalizeHostLike(String value) {
|
||||
if (value == null) return null;
|
||||
String raw = value.trim();
|
||||
if (raw.isEmpty()) return null;
|
||||
try {
|
||||
String withScheme = raw.matches("^[a-zA-Z]+://.*$") ? raw : "https://" + raw;
|
||||
URI uri = URI.create(withScheme);
|
||||
String host = uri.getHost();
|
||||
if (host == null || host.isBlank()) return null;
|
||||
return host.trim().toLowerCase(Locale.ROOT);
|
||||
} catch (Exception e) {
|
||||
String cleaned = raw
|
||||
.replaceFirst("^[a-zA-Z]+://", "")
|
||||
.replaceFirst("/.*$", "")
|
||||
.trim()
|
||||
.toLowerCase(Locale.ROOT);
|
||||
return cleaned.isEmpty() ? null : cleaned;
|
||||
}
|
||||
}
|
||||
|
||||
private static String toHex(byte[] bytes) {
|
||||
if (bytes == null) return "";
|
||||
StringBuilder sb = new StringBuilder(bytes.length * 2);
|
||||
for (byte b : bytes) {
|
||||
sb.append(Character.forDigit((b >>> 4) & 0xF, 16));
|
||||
sb.append(Character.forDigit(b & 0xF, 16));
|
||||
}
|
||||
return sb.toString();
|
||||
}
|
||||
|
||||
private static void tryAbort(WebSocket webSocket) {
|
||||
try {
|
||||
webSocket.sendClose(WebSocket.NORMAL_CLOSURE, "ok");
|
||||
} catch (Exception ignored) {
|
||||
}
|
||||
try {
|
||||
webSocket.abort();
|
||||
} catch (Exception ignored) {
|
||||
}
|
||||
}
|
||||
|
||||
private record AddBlockPushResult(
|
||||
boolean ok,
|
||||
int status,
|
||||
String code,
|
||||
int serverLastGlobalNumber,
|
||||
String serverLastGlobalHash
|
||||
) {
|
||||
static AddBlockPushResult success() {
|
||||
return new AddBlockPushResult(true, 200, "", Integer.MIN_VALUE, "");
|
||||
}
|
||||
|
||||
boolean needsBackfill() {
|
||||
return !ok && ("bad_prev_hash".equalsIgnoreCase(code) || "bad_block_number".equalsIgnoreCase(code));
|
||||
}
|
||||
}
|
||||
|
||||
private static final class SyncWsListener implements WebSocket.Listener {
|
||||
private final CompletableFuture<String> responseFuture;
|
||||
private final CountDownLatch openLatch;
|
||||
private final StringBuilder textBuffer = new StringBuilder();
|
||||
|
||||
private SyncWsListener(CompletableFuture<String> responseFuture, CountDownLatch openLatch) {
|
||||
this.responseFuture = responseFuture;
|
||||
this.openLatch = openLatch;
|
||||
}
|
||||
|
||||
@Override
|
||||
public void onOpen(WebSocket webSocket) {
|
||||
openLatch.countDown();
|
||||
webSocket.request(1);
|
||||
}
|
||||
|
||||
@Override
|
||||
public CompletionStage<?> onText(WebSocket webSocket, CharSequence data, boolean last) {
|
||||
textBuffer.append(data);
|
||||
if (last && !responseFuture.isDone()) {
|
||||
responseFuture.complete(textBuffer.toString());
|
||||
}
|
||||
webSocket.request(1);
|
||||
return CompletableFuture.completedFuture(null);
|
||||
}
|
||||
|
||||
@Override
|
||||
public CompletionStage<?> onBinary(WebSocket webSocket, ByteBuffer data, boolean last) {
|
||||
webSocket.request(1);
|
||||
return CompletableFuture.completedFuture(null);
|
||||
}
|
||||
|
||||
@Override
|
||||
public CompletionStage<?> onClose(WebSocket webSocket, int statusCode, String reason) {
|
||||
if (!responseFuture.isDone()) {
|
||||
responseFuture.completeExceptionally(new IllegalStateException("WS closed before response: " + statusCode + " " + reason));
|
||||
}
|
||||
return CompletableFuture.completedFuture(null);
|
||||
}
|
||||
|
||||
@Override
|
||||
public void onError(WebSocket webSocket, Throwable error) {
|
||||
if (!responseFuture.isDone()) {
|
||||
responseFuture.completeExceptionally(error);
|
||||
}
|
||||
openLatch.countDown();
|
||||
}
|
||||
}
|
||||
}
|
||||
Reference in New Issue
Block a user