diff --git a/SHiNE-server/shine-server-db/src/main/java/shine/db/dao/UserAccessServersCurrentDAO.java b/SHiNE-server/shine-server-db/src/main/java/shine/db/dao/UserAccessServersCurrentDAO.java index e95a6df4..c53d0120 100644 --- a/SHiNE-server/shine-server-db/src/main/java/shine/db/dao/UserAccessServersCurrentDAO.java +++ b/SHiNE-server/shine-server-db/src/main/java/shine/db/dao/UserAccessServersCurrentDAO.java @@ -75,6 +75,35 @@ public final class UserAccessServersCurrentDAO { return result; } + /** + * Возвращает уникальные физические серверы из текущей routing-проекции. + * Используется транспортным пулом: одно WSS-соединение держится на сервер, + * а не на каждого пользователя этого сервера. + */ + public List listDistinctServers() throws SQLException { + String sql = """ + SELECT DISTINCT ON (LOWER(server_login)) + server_login, server_url + FROM user_access_servers_current + WHERE server_login IS NOT NULL AND BTRIM(server_login) <> '' + AND server_url IS NOT NULL AND BTRIM(server_url) <> '' + ORDER BY LOWER(server_login), server_login, server_url + """; + List result = new ArrayList<>(); + try (Connection c = db.getConnection(); + PreparedStatement ps = c.prepareStatement(sql); + ResultSet rs = ps.executeQuery()) { + while (rs.next()) { + UserAccessServerRouteEntry entry = new UserAccessServerRouteEntry(); + entry.setUserLogin(""); + entry.setServerLogin(rs.getString("server_login")); + entry.setServerUrl(rs.getString("server_url")); + result.add(entry); + } + } + return result; + } + private UserAccessServerRouteEntry mapRow(ResultSet rs) throws SQLException { UserAccessServerRouteEntry entry = new UserAccessServerRouteEntry(); entry.setUserLogin(rs.getString("user_login")); diff --git a/SHiNE-server/shine-server-net-protocol/src/main/java/server/logic/ws_protocol/JSON/ConnectionContext.java b/SHiNE-server/shine-server-net-protocol/src/main/java/server/logic/ws_protocol/JSON/ConnectionContext.java index 4877f45f..37bf1a9a 100644 --- a/SHiNE-server/shine-server-net-protocol/src/main/java/server/logic/ws_protocol/JSON/ConnectionContext.java +++ b/SHiNE-server/shine-server-net-protocol/src/main/java/server/logic/ws_protocol/JSON/ConnectionContext.java @@ -4,6 +4,8 @@ import org.eclipse.jetty.websocket.api.Session; import shine.db.entities.CurrentUserEntry; import shine.db.entities.ActiveSessionEntry; +import java.util.List; + /** * ConnectionContext — контекст состояния одного WebSocket-соединения. * Живёт ровно столько же, сколько живёт подключение. @@ -77,6 +79,12 @@ public class ConnectionContext { */ private Session wsSession; + /** Временная server-to-server роль, заявленная через ServerHello. */ + private boolean serverConnection; + private String remoteServerLogin; + private int remoteServerProtocolVersion; + private List remoteServerCapabilities = List.of(); + // --- WebSocket Session --- public Session getWsSession() { @@ -87,6 +95,20 @@ public class ConnectionContext { this.wsSession = wsSession; } + public boolean isServerConnection() { return serverConnection; } + public void setServerConnection(boolean serverConnection) { this.serverConnection = serverConnection; } + + public String getRemoteServerLogin() { return remoteServerLogin; } + public void setRemoteServerLogin(String remoteServerLogin) { this.remoteServerLogin = remoteServerLogin; } + + public int getRemoteServerProtocolVersion() { return remoteServerProtocolVersion; } + public void setRemoteServerProtocolVersion(int value) { this.remoteServerProtocolVersion = value; } + + public List getRemoteServerCapabilities() { return remoteServerCapabilities; } + public void setRemoteServerCapabilities(List capabilities) { + this.remoteServerCapabilities = capabilities == null ? List.of() : List.copyOf(capabilities); + } + // --- SolanaUser / ActiveSession --- public CurrentUserEntry getCurrentUser() { @@ -188,6 +210,10 @@ public class ConnectionContext { authenticationStatus = AUTH_STATUS_NONE; wsSession = null; + serverConnection = false; + remoteServerLogin = null; + remoteServerProtocolVersion = 0; + remoteServerCapabilities = List.of(); } @Override diff --git a/SHiNE-server/shine-server-net-protocol/src/main/java/server/logic/ws_protocol/JSON/JsonHandlerRegistry.java b/SHiNE-server/shine-server-net-protocol/src/main/java/server/logic/ws_protocol/JSON/JsonHandlerRegistry.java index 336aaf5f..a7b5260a 100644 --- a/SHiNE-server/shine-server-net-protocol/src/main/java/server/logic/ws_protocol/JSON/JsonHandlerRegistry.java +++ b/SHiNE-server/shine-server-net-protocol/src/main/java/server/logic/ws_protocol/JSON/JsonHandlerRegistry.java @@ -133,6 +133,7 @@ import server.logic.ws_protocol.JSON.handlers.system.Net_ClientDebugLog_Handler; import server.logic.ws_protocol.JSON.handlers.system.Net_ListBlockchainHeads_Handler; import server.logic.ws_protocol.JSON.handlers.system.Net_CallDeliveryReport_Handler; import server.logic.ws_protocol.JSON.handlers.system.Net_Ping_Handler; +import server.logic.ws_protocol.JSON.handlers.system.Net_ServerHello_Handler; import server.logic.ws_protocol.JSON.handlers.system.entyties.Net_CallDeliveryReport_Request; import server.logic.ws_protocol.JSON.handlers.system.entyties.Net_ClientErrorLog_Request; import server.logic.ws_protocol.JSON.handlers.system.entyties.Net_ClientDebugLog_Request; @@ -141,6 +142,7 @@ import server.logic.ws_protocol.JSON.handlers.system.entyties.Net_GetServerInfo_ import server.logic.ws_protocol.JSON.handlers.system.entyties.Net_GetSyncUserProfile_Request; import server.logic.ws_protocol.JSON.handlers.system.entyties.Net_ListBlockchainHeads_Request; import server.logic.ws_protocol.JSON.handlers.system.entyties.Net_Ping_Request; +import server.logic.ws_protocol.JSON.handlers.system.entyties.Net_ServerHello_Request; import java.util.Map; @@ -232,6 +234,7 @@ public final class JsonHandlerRegistry { Map.entry("SendSignal", new Net_SendSignal_Handler()), // --- system --- + Map.entry("ServerHello", new Net_ServerHello_Handler()), Map.entry("Ping", new Net_Ping_Handler()), Map.entry("GetServerInfo", new Net_GetServerInfo_Handler()), Map.entry("ListBlockchainHeads", new Net_ListBlockchainHeads_Handler()), @@ -323,6 +326,7 @@ public final class JsonHandlerRegistry { Map.entry("SendSignal", Net_SendSignal_Request.class), // --- system --- + Map.entry("ServerHello", Net_ServerHello_Request.class), Map.entry("Ping", Net_Ping_Request.class), Map.entry("GetServerInfo", Net_GetServerInfo_Request.class), Map.entry("ListBlockchainHeads", Net_ListBlockchainHeads_Request.class), diff --git a/SHiNE-server/shine-server-net-protocol/src/main/java/server/logic/ws_protocol/JSON/handlers/userSettings/Net_UpsertUserSetting_Handler.java b/SHiNE-server/shine-server-net-protocol/src/main/java/server/logic/ws_protocol/JSON/handlers/userSettings/Net_UpsertUserSetting_Handler.java index 0a9fff3f..c30a6b32 100644 --- a/SHiNE-server/shine-server-net-protocol/src/main/java/server/logic/ws_protocol/JSON/handlers/userSettings/Net_UpsertUserSetting_Handler.java +++ b/SHiNE-server/shine-server-net-protocol/src/main/java/server/logic/ws_protocol/JSON/handlers/userSettings/Net_UpsertUserSetting_Handler.java @@ -123,7 +123,7 @@ public class Net_UpsertUserSetting_Handler implements JsonMessageHandler { if (remoteLogin.isBlank() || remoteUrl.isBlank()) continue; if (!ownServerLogin.isBlank() && remoteLogin.equalsIgnoreCase(ownServerLogin)) continue; try { - REMOTE.upsertUserSetting(remoteUrl, entry, true); + REMOTE.upsertUserSetting(remoteLogin, remoteUrl, entry, true); delivered++; } catch (Exception e) { log.warn("user_settings immediate sync failed: login={} remoteServer={} reason={}", login, remoteLogin, String.valueOf(e)); diff --git a/SHiNE-server/shine-server-net-protocol/src/main/java/server/sync/AddBlockSyncService.java b/SHiNE-server/shine-server-net-protocol/src/main/java/server/sync/AddBlockSyncService.java index 2e4741ab..f7d35acc 100644 --- a/SHiNE-server/shine-server-net-protocol/src/main/java/server/sync/AddBlockSyncService.java +++ b/SHiNE-server/shine-server-net-protocol/src/main/java/server/sync/AddBlockSyncService.java @@ -11,27 +11,19 @@ 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. + * Фоновая репликация AddBlock через общий постоянный WSS-пул на серверы + * из локальной таблицы 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()), @@ -100,14 +92,14 @@ public final class AddBlockSyncService { } private void replicateToPartner(SyncServerEntry partner, String blockchainName, int blockNumber, BlockEntry currentBlock) throws Exception { - String wsUrl = buildWsUrl(partner.getServerAddress()); + String wsUrl = ServerConnectionPool.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); + AddBlockPushResult firstTry = pushBlock(partner, blockchainName, currentBlock); if (firstTry.ok()) { log.info("AddBlock sync ok: partner={} blockchainName={} blockNumber={}", partner.getLogin(), blockchainName, blockNumber); @@ -142,7 +134,7 @@ public final class AddBlockSyncService { } for (BlockEntry blockEntry : missingBlocks) { - AddBlockPushResult backfillResult = pushBlock(wsUrl, blockchainName, blockEntry); + AddBlockPushResult backfillResult = pushBlock(partner, blockchainName, blockEntry); if (!backfillResult.ok()) { log.warn("AddBlock sync backfill failed: partner={} blockchainName={} blockNumber={} code={}", partner.getLogin(), blockchainName, blockEntry.getBlockNumber(), backfillResult.code()); @@ -154,8 +146,8 @@ public final class AddBlockSyncService { partner.getLogin(), blockchainName, fromBlockNumber, blockNumber); } - private AddBlockPushResult pushBlock(String wsUrl, String blockchainName, BlockEntry blockEntry) throws Exception { - JsonNode response = sendAddBlock(wsUrl, blockchainName, blockEntry); + private AddBlockPushResult pushBlock(SyncServerEntry partner, String blockchainName, BlockEntry blockEntry) throws Exception { + JsonNode response = sendAddBlock(partner, blockchainName, blockEntry); int status = response.path("status").asInt(500); if (status >= 200 && status < 300) { return AddBlockPushResult.success(); @@ -172,31 +164,16 @@ public final class AddBlockSyncService { return new AddBlockPushResult(false, status, code, serverLastGlobalNumber, serverLastGlobalHash); } - private JsonNode sendAddBlock(String wsUrl, String blockchainName, BlockEntry blockEntry) throws Exception { - CompletableFuture 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 JsonNode sendAddBlock(SyncServerEntry partner, String blockchainName, BlockEntry blockEntry) throws Exception { + String jsonTemplate = buildAddBlockJsonTemplate(blockchainName, blockEntry); + return ServerConnectionPool.getInstance().request( + partner.getLogin(), + partner.getServerAddress(), + jsonTemplate, + ServerConnectionPool.Priority.BULK); } - private String buildAddBlockJson(String requestId, String blockchainName, BlockEntry blockEntry) throws Exception { + private String buildAddBlockJsonTemplate(String blockchainName, BlockEntry blockEntry) throws Exception { String prevHashHex = blockEntry.getBlockNumber() <= 0 ? "" : toHex(extractPrevHash32(blockEntry.getBlockBytes())); @@ -205,8 +182,6 @@ public final class AddBlockSyncService { String safeBlockchainName = MAPPER.writeValueAsString(blockchainName); String safePrevHashHex = MAPPER.writeValueAsString(prevHashHex); String safeBlockBytes = MAPPER.writeValueAsString(blockBytesB64); - String safeRequestId = MAPPER.writeValueAsString(requestId); - return """ { "op":"AddBlock", @@ -218,7 +193,7 @@ public final class AddBlockSyncService { "blockBytesB64":%s } } - """.formatted(safeRequestId, safeBlockchainName, blockEntry.getBlockNumber(), safePrevHashHex, safeBlockBytes); + """.formatted("%s", safeBlockchainName, blockEntry.getBlockNumber(), safePrevHashHex, safeBlockBytes); } private static byte[] extractPrevHash32(byte[] blockBytes) { @@ -240,32 +215,6 @@ public final class AddBlockSyncService { 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); @@ -276,17 +225,6 @@ public final class AddBlockSyncService { 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, @@ -309,52 +247,4 @@ public final class AddBlockSyncService { } } - private static final class SyncWsListener implements WebSocket.Listener { - private final CompletableFuture responseFuture; - private final CountDownLatch openLatch; - private final StringBuilder textBuffer = new StringBuilder(); - - private SyncWsListener(CompletableFuture 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(); - } - } } diff --git a/SHiNE-server/shine-server-net-protocol/src/main/java/server/sync/DmDeliveryCoordinator.java b/SHiNE-server/shine-server-net-protocol/src/main/java/server/sync/DmDeliveryCoordinator.java index 4b3223e0..4d9577f8 100644 --- a/SHiNE-server/shine-server-net-protocol/src/main/java/server/sync/DmDeliveryCoordinator.java +++ b/SHiNE-server/shine-server-net-protocol/src/main/java/server/sync/DmDeliveryCoordinator.java @@ -149,7 +149,8 @@ public final class DmDeliveryCoordinator { completion.submit(() -> { try { if (!routeLogin.equals(ownServerLogin)) { - REMOTE.receiveIncomingMessage(route.getServerUrl(), incomingBlobB64, ownServerLogin); + REMOTE.receiveIncomingMessage( + routeLogin, route.getServerUrl(), incomingBlobB64, ownServerLogin); } return new RouteAttempt(routeLogin, null); } catch (Exception e) { @@ -188,6 +189,7 @@ public final class DmDeliveryCoordinator { if (incoming == null || outgoing == null) return current; try { REMOTE.sendMessagePair( + peer.getServerLogin(), peer.getServerUrl(), Base64.getEncoder().encodeToString(incoming.getRawBlock()), Base64.getEncoder().encodeToString(outgoing.getRawBlock()), @@ -204,7 +206,7 @@ public final class DmDeliveryCoordinator { if (peer == null) return current; try { RemoteDmSyncClient.RemoteDeliveryStatus remote = REMOTE.getDmDeliveryStatus( - peer.getServerUrl(), current.getOutgoingMessageKey()); + peer.getServerLogin(), peer.getServerUrl(), current.getOutgoingMessageKey()); if (remote.known() && remote.delivered()) { return DELIVERY_DAO.markDeliveredFromPeer(current.getEventId(), System.currentTimeMillis()); } diff --git a/SHiNE-server/shine-server-net-protocol/src/main/java/server/sync/DmFederationService.java b/SHiNE-server/shine-server-net-protocol/src/main/java/server/sync/DmFederationService.java index 47994619..845a8778 100644 --- a/SHiNE-server/shine-server-net-protocol/src/main/java/server/sync/DmFederationService.java +++ b/SHiNE-server/shine-server-net-protocol/src/main/java/server/sync/DmFederationService.java @@ -27,12 +27,15 @@ public final class DmFederationService { String ownServerLogin = ownServerLogin(); for (UserAccessServerRouteEntry route : senderRoutes.values()) { if (isOwnServer(route, ownServerLogin)) continue; - REMOTE.sendMessagePair(route.getServerUrl(), incomingBlobB64, outgoingBlobB64, ownServerLogin); + REMOTE.sendMessagePair( + route.getServerLogin(), route.getServerUrl(), + incomingBlobB64, outgoingBlobB64, ownServerLogin); } for (UserAccessServerRouteEntry route : recipientRoutes.values()) { if (isOwnServer(route, ownServerLogin)) continue; if (senderRoutes.containsKey(normalize(route.getServerLogin()))) continue; - REMOTE.receiveIncomingMessage(route.getServerUrl(), incomingBlobB64, ownServerLogin); + REMOTE.receiveIncomingMessage( + route.getServerLogin(), route.getServerUrl(), incomingBlobB64, ownServerLogin); } } catch (Exception e) { log.warn("DM federation pair fan-out failed: from={} to={}", fromLogin, toLogin, e); @@ -54,7 +57,8 @@ public final class DmFederationService { if (routeLogin == null) continue; if (ownServerLogin != null && ownServerLogin.equalsIgnoreCase(routeLogin)) continue; if (normalizedSource != null && normalizedSource.equalsIgnoreCase(routeLogin)) continue; - REMOTE.receiveIncomingMessage(route.getServerUrl(), incomingBlobB64, ownServerLogin); + REMOTE.receiveIncomingMessage( + route.getServerLogin(), route.getServerUrl(), incomingBlobB64, ownServerLogin); } } catch (Exception e) { log.warn("DM federation incoming relay failed: to={}", toLogin, e); @@ -79,9 +83,9 @@ public final class DmFederationService { for (UserAccessServerRouteEntry route : routes.values()) { if (isOwnServer(route, ownServerLogin)) continue; if (oneMessageDelete) { - REMOTE.deleteMessage(route.getServerUrl(), blobB64); + REMOTE.deleteMessage(route.getServerLogin(), route.getServerUrl(), blobB64); } else { - REMOTE.deleteConversation(route.getServerUrl(), blobB64); + REMOTE.deleteConversation(route.getServerLogin(), route.getServerUrl(), blobB64); } } } catch (Exception e) { diff --git a/SHiNE-server/shine-server-net-protocol/src/main/java/server/sync/RemoteBlockchainSyncClient.java b/SHiNE-server/shine-server-net-protocol/src/main/java/server/sync/RemoteBlockchainSyncClient.java index 1ac52c97..7d4e4e87 100644 --- a/SHiNE-server/shine-server-net-protocol/src/main/java/server/sync/RemoteBlockchainSyncClient.java +++ b/SHiNE-server/shine-server-net-protocol/src/main/java/server/sync/RemoteBlockchainSyncClient.java @@ -2,37 +2,22 @@ package server.sync; import com.fasterxml.jackson.databind.JsonNode; import com.fasterxml.jackson.databind.ObjectMapper; -import org.slf4j.Logger; -import org.slf4j.LoggerFactory; - -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.ArrayList; import java.util.List; -import java.util.Locale; -import java.util.UUID; -import java.util.concurrent.CompletableFuture; -import java.util.concurrent.CompletionStage; -import java.util.concurrent.CountDownLatch; -import java.util.concurrent.TimeUnit; -import java.util.concurrent.TimeoutException; /** * Минимальный клиент для межсерверных JSON-op запросов по WSS. */ public final class RemoteBlockchainSyncClient { - private static final Logger log = LoggerFactory.getLogger(RemoteBlockchainSyncClient.class); private static final ObjectMapper MAPPER = new ObjectMapper(); - private static final HttpClient HTTP = HttpClient.newBuilder() - .connectTimeout(Duration.ofSeconds(6)) - .build(); public List listBlockchainHeads(String serverAddressRaw) throws Exception { - JsonNode response = send(serverAddressRaw, """ + return listBlockchainHeads(null, serverAddressRaw); + } + + public List listBlockchainHeads(String serverLogin, String serverAddressRaw) throws Exception { + JsonNode response = send(serverLogin, serverAddressRaw, """ { "op":"ListBlockchainHeads", "requestId":%s, @@ -62,8 +47,12 @@ public final class RemoteBlockchainSyncClient { } public RemoteSyncUserProfile getSyncUserProfile(String serverAddressRaw, String login) throws Exception { + return getSyncUserProfile(null, serverAddressRaw, login); + } + + public RemoteSyncUserProfile getSyncUserProfile(String serverLogin, String serverAddressRaw, String login) throws Exception { String safeLogin = MAPPER.writeValueAsString(login); - JsonNode response = send(serverAddressRaw, """ + JsonNode response = send(serverLogin, serverAddressRaw, """ { "op":"GetSyncUserProfile", "requestId":%s, @@ -96,9 +85,17 @@ public final class RemoteBlockchainSyncClient { ); } - public RemoteBlockchainBlock getBlockchainBlock(String serverAddressRaw, String blockchainName, int blockNumber) throws Exception { + public RemoteBlockchainBlock getBlockchainBlock( + String serverAddressRaw, String blockchainName, int blockNumber + ) throws Exception { + return getBlockchainBlock(null, serverAddressRaw, blockchainName, blockNumber); + } + + public RemoteBlockchainBlock getBlockchainBlock( + String serverLogin, String serverAddressRaw, String blockchainName, int blockNumber + ) throws Exception { String safeBlockchainName = MAPPER.writeValueAsString(blockchainName); - JsonNode response = send(serverAddressRaw, """ + JsonNode response = send(serverLogin, serverAddressRaw, """ { "op":"GetBlockchainBlock", "requestId":%s, @@ -126,32 +123,12 @@ public final class RemoteBlockchainSyncClient { ); } - private JsonNode send(String serverAddressRaw, String jsonTemplate) throws Exception { - String requestId = MAPPER.writeValueAsString("sync-" + UUID.randomUUID()); - String json = jsonTemplate.formatted(requestId); - String wsUrl = buildWsUrl(serverAddressRaw); - if (wsUrl == null) { - throw new IllegalArgumentException("Invalid server address: " + serverAddressRaw); - } - - CompletableFuture 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"); - } - - webSocket.sendText(json, true).get(8, TimeUnit.SECONDS); - String responseJson = responseFuture.get(12, TimeUnit.SECONDS); - tryAbort(webSocket); - return MAPPER.readTree(responseJson); + private JsonNode send(String serverLogin, String serverAddressRaw, String jsonTemplate) throws Exception { + return ServerConnectionPool.getInstance().request( + serverLogin, + serverAddressRaw, + jsonTemplate, + ServerConnectionPool.Priority.BULK); } private static String errorCode(JsonNode response) { @@ -161,40 +138,7 @@ public final class RemoteBlockchainSyncClient { } 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 void tryAbort(WebSocket webSocket) { - try { - webSocket.sendClose(WebSocket.NORMAL_CLOSURE, "ok"); - } catch (Exception ignored) { - } - try { - webSocket.abort(); - } catch (Exception ignored) { - } + return ServerConnectionPool.buildWsUrl(serverAddressRaw); } public record RemoteBlockchainHead( @@ -220,53 +164,4 @@ public final class RemoteBlockchainSyncClient { long blockchainSizeLimitBytes ) {} - private static final class SyncWsListener implements WebSocket.Listener { - private final CompletableFuture responseFuture; - private final CountDownLatch openLatch; - private final StringBuilder textBuffer = new StringBuilder(); - - private SyncWsListener(CompletableFuture 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) { - log.warn("Remote sync websocket error: {}", String.valueOf(error)); - if (!responseFuture.isDone()) { - responseFuture.completeExceptionally(error); - } - openLatch.countDown(); - } - } } diff --git a/SHiNE-server/shine-server-net-protocol/src/main/java/server/sync/RemoteDmSyncClient.java b/SHiNE-server/shine-server-net-protocol/src/main/java/server/sync/RemoteDmSyncClient.java index a0071818..16b1280f 100644 --- a/SHiNE-server/shine-server-net-protocol/src/main/java/server/sync/RemoteDmSyncClient.java +++ b/SHiNE-server/shine-server-net-protocol/src/main/java/server/sync/RemoteDmSyncClient.java @@ -3,37 +3,33 @@ package server.sync; import com.fasterxml.jackson.databind.JsonNode; import com.fasterxml.jackson.databind.ObjectMapper; -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.ArrayList; import java.util.List; -import java.util.UUID; -import java.util.concurrent.CompletableFuture; -import java.util.concurrent.CompletionStage; -import java.util.concurrent.CountDownLatch; -import java.util.concurrent.TimeUnit; -import java.util.concurrent.TimeoutException; /** Клиент стабильных межсерверных DM-операций. */ public final class RemoteDmSyncClient { private static final ObjectMapper MAPPER = new ObjectMapper(); - private static final HttpClient HTTP = HttpClient.newBuilder() - .connectTimeout(Duration.ofSeconds(6)) - .build(); public void sendMessagePair( String serverAddressRaw, String incomingBlobB64, String outgoingBlobB64, String sourceServerLogin + ) throws Exception { + sendMessagePair(null, serverAddressRaw, incomingBlobB64, outgoingBlobB64, sourceServerLogin); + } + + public void sendMessagePair( + String targetServerLogin, + String serverAddressRaw, + String incomingBlobB64, + String outgoingBlobB64, + String sourceServerLogin ) throws Exception { String incomingJson = MAPPER.writeValueAsString(incomingBlobB64); String outgoingJson = MAPPER.writeValueAsString(outgoingBlobB64); String sourceServerLoginJson = toOptionalJsonField("sourceServerLogin", sourceServerLogin); - JsonNode response = send(serverAddressRaw, """ + JsonNode response = send(targetServerLogin, serverAddressRaw, """ { "op":"ReceiveOutcomingMessage", "requestId":%s, @@ -50,10 +46,19 @@ public final class RemoteDmSyncClient { String serverAddressRaw, String incomingBlobB64, String sourceServerLogin + ) throws Exception { + receiveIncomingMessage(null, serverAddressRaw, incomingBlobB64, sourceServerLogin); + } + + public void receiveIncomingMessage( + String targetServerLogin, + String serverAddressRaw, + String incomingBlobB64, + String sourceServerLogin ) throws Exception { String incomingJson = MAPPER.writeValueAsString(incomingBlobB64); String sourceServerLoginJson = toOptionalJsonField("sourceServerLogin", sourceServerLogin); - JsonNode response = send(serverAddressRaw, """ + JsonNode response = send(targetServerLogin, serverAddressRaw, """ { "op":"ReceiveIncomingMessage", "requestId":%s, @@ -66,8 +71,14 @@ public final class RemoteDmSyncClient { } public RemoteDeliveryStatus getDmDeliveryStatus(String serverAddressRaw, String messageKey) throws Exception { + return getDmDeliveryStatus(null, serverAddressRaw, messageKey); + } + + public RemoteDeliveryStatus getDmDeliveryStatus( + String targetServerLogin, String serverAddressRaw, String messageKey + ) throws Exception { String messageKeyJson = MAPPER.writeValueAsString(messageKey); - JsonNode response = send(serverAddressRaw, """ + JsonNode response = send(targetServerLogin, serverAddressRaw, """ { "op":"GetDmDeliveryStatus", "requestId":%s, @@ -94,7 +105,21 @@ public final class RemoteDmSyncClient { int maxBytes, List ackSyncIds ) throws Exception { - try (RemoteSyncSession session = new RemoteSyncSession(serverAddressRaw)) { + return dmSyncBatch(null, serverAddressRaw, ownerLogin, afterStoredAtMs, + afterMessageKey, limit, maxBytes, ackSyncIds); + } + + public RemoteDmBatch dmSyncBatch( + String targetServerLogin, + String serverAddressRaw, + String ownerLogin, + long afterStoredAtMs, + String afterMessageKey, + int limit, + int maxBytes, + List ackSyncIds + ) throws Exception { + try (RemoteSyncSession session = new RemoteSyncSession(targetServerLogin, serverAddressRaw)) { return dmSyncBatch(session, ownerLogin, afterStoredAtMs, afterMessageKey, limit, maxBytes, ackSyncIds); } @@ -156,8 +181,12 @@ public final class RemoteDmSyncClient { } public void deleteMessage(String serverAddressRaw, String blobB64) throws Exception { + deleteMessage(null, serverAddressRaw, blobB64); + } + + public void deleteMessage(String targetServerLogin, String serverAddressRaw, String blobB64) throws Exception { String blobJson = MAPPER.writeValueAsString(blobB64); - JsonNode response = send(serverAddressRaw, """ + JsonNode response = send(targetServerLogin, serverAddressRaw, """ { "op":"DeleteMessage", "requestId":%s, @@ -168,8 +197,12 @@ public final class RemoteDmSyncClient { } public void deleteConversation(String serverAddressRaw, String blobB64) throws Exception { + deleteConversation(null, serverAddressRaw, blobB64); + } + + public void deleteConversation(String targetServerLogin, String serverAddressRaw, String blobB64) throws Exception { String blobJson = MAPPER.writeValueAsString(blobB64); - JsonNode response = send(serverAddressRaw, """ + JsonNode response = send(targetServerLogin, serverAddressRaw, """ { "op":"DeleteConversation", "requestId":%s, @@ -179,27 +212,12 @@ public final class RemoteDmSyncClient { ensureOk("DeleteConversation", response); } - private JsonNode send(String serverAddressRaw, String jsonTemplate) throws Exception { - String requestId = MAPPER.writeValueAsString("dm-sync-" + UUID.randomUUID()); - String json = jsonTemplate.formatted(requestId); - String wsUrl = RemoteBlockchainSyncClient.buildWsUrl(serverAddressRaw); - if (wsUrl == null) throw new IllegalArgumentException("Invalid server address: " + serverAddressRaw); - - CompletableFuture 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"); - } - webSocket.sendText(json, true).get(8, TimeUnit.SECONDS); - String responseJson = responseFuture.get(12, TimeUnit.SECONDS); - tryAbort(webSocket); - return MAPPER.readTree(responseJson); + private JsonNode send(String targetServerLogin, String serverAddressRaw, String jsonTemplate) throws Exception { + return ServerConnectionPool.getInstance().request( + targetServerLogin, + serverAddressRaw, + jsonTemplate, + ServerConnectionPool.Priority.REALTIME); } private String toOptionalJsonField(String fieldName, String value) throws Exception { @@ -215,60 +233,10 @@ public final class RemoteDmSyncClient { throw new IllegalStateException(op + " failed: status=" + status + " code=" + code); } - private static void tryAbort(WebSocket webSocket) { - try { webSocket.sendClose(WebSocket.NORMAL_CLOSURE, "ok"); } catch (Exception ignored) {} - try { webSocket.abort(); } catch (Exception ignored) {} - } - public record RemoteDeliveryStatus(String messageKey, boolean known, boolean delivered) {} public record RemoteDmBatch(long nextStoredAtMs, String nextMessageKey, boolean hasMore, List items) {} public record RemoteDmItem( String syncId, String primaryMessageKey, long storedAtMs, List blobsB64 ) {} - private static final class SyncWsListener implements WebSocket.Listener { - private final CompletableFuture responseFuture; - private final CountDownLatch openLatch; - private final StringBuilder textBuffer = new StringBuilder(); - - private SyncWsListener(CompletableFuture 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(); - } - } } diff --git a/SHiNE-server/shine-server-net-protocol/src/main/java/server/sync/RemoteSyncSession.java b/SHiNE-server/shine-server-net-protocol/src/main/java/server/sync/RemoteSyncSession.java index a3a58691..3f5bca1f 100644 --- a/SHiNE-server/shine-server-net-protocol/src/main/java/server/sync/RemoteSyncSession.java +++ b/SHiNE-server/shine-server-net-protocol/src/main/java/server/sync/RemoteSyncSession.java @@ -1,70 +1,35 @@ package server.sync; import com.fasterxml.jackson.databind.JsonNode; -import com.fasterxml.jackson.databind.ObjectMapper; -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.UUID; -import java.util.concurrent.CompletableFuture; -import java.util.concurrent.CompletionStage; -import java.util.concurrent.LinkedBlockingQueue; -import java.util.concurrent.TimeUnit; -import java.util.concurrent.TimeoutException; - -/** Один последовательный WS-сеанс для синхронизации всех данных access-сервера. */ +/** + * Логический последовательный сеанс поверх общего постоянного WSS-пула. + * close() больше не закрывает физическое соединение с сервером. + */ public final class RemoteSyncSession implements AutoCloseable { - private static final ObjectMapper MAPPER = new ObjectMapper(); - private static final HttpClient HTTP = HttpClient.newBuilder() - .connectTimeout(Duration.ofSeconds(6)).build(); - private final LinkedBlockingQueue responses = new LinkedBlockingQueue<>(); - private final WebSocket webSocket; + private final String serverLogin; + private final String serverAddress; + /** Совместимый конструктор: при отсутствии логина пул использует адрес как ключ peer. */ public RemoteSyncSession(String serverAddressRaw) throws Exception { - String wsUrl = RemoteBlockchainSyncClient.buildWsUrl(serverAddressRaw); - if (wsUrl == null) throw new IllegalArgumentException("Invalid server address: " + serverAddressRaw); - Listener listener = new Listener(responses); - webSocket = HTTP.newWebSocketBuilder().connectTimeout(Duration.ofSeconds(6)) - .buildAsync(URI.create(wsUrl), listener).get(8, TimeUnit.SECONDS); + this(null, serverAddressRaw); + } + + public RemoteSyncSession(String serverLogin, String serverAddressRaw) { + this.serverLogin = serverLogin; + this.serverAddress = serverAddressRaw; } public synchronized JsonNode send(String jsonTemplate) throws Exception { - String requestId = MAPPER.writeValueAsString("access-sync-" + UUID.randomUUID()); - webSocket.sendText(jsonTemplate.formatted(requestId), true).get(8, TimeUnit.SECONDS); - String json = responses.poll(12, TimeUnit.SECONDS); - if (json == null) throw new TimeoutException("WS response timeout"); - return MAPPER.readTree(json); + return ServerConnectionPool.getInstance().request( + serverLogin, + serverAddress, + jsonTemplate, + ServerConnectionPool.Priority.NORMAL); } @Override public void close() { - try { webSocket.sendClose(WebSocket.NORMAL_CLOSURE, "ok"); } catch (Exception ignored) {} - } - - private static final class Listener implements WebSocket.Listener { - private final LinkedBlockingQueue responses; - private final StringBuilder text = new StringBuilder(); - - private Listener(LinkedBlockingQueue responses) { this.responses = responses; } - @Override public void onOpen(WebSocket ws) { ws.request(1); } - @Override public CompletionStage onText(WebSocket ws, CharSequence data, boolean last) { - text.append(data); - if (last) { - responses.offer(text.toString()); - text.setLength(0); - } - ws.request(1); - return CompletableFuture.completedFuture(null); - } - @Override public CompletionStage onBinary(WebSocket ws, ByteBuffer data, boolean last) { - ws.request(1); - return CompletableFuture.completedFuture(null); - } - @Override public void onError(WebSocket ws, Throwable error) { - responses.offer("{\"status\":500,\"code\":\"WS_ERROR\"}"); - } + // Физическое соединение принадлежит ServerConnectionPool. } } diff --git a/SHiNE-server/shine-server-net-protocol/src/main/java/server/sync/RemoteUserSettingsSyncClient.java b/SHiNE-server/shine-server-net-protocol/src/main/java/server/sync/RemoteUserSettingsSyncClient.java index 3871acab..08654cf9 100644 --- a/SHiNE-server/shine-server-net-protocol/src/main/java/server/sync/RemoteUserSettingsSyncClient.java +++ b/SHiNE-server/shine-server-net-protocol/src/main/java/server/sync/RemoteUserSettingsSyncClient.java @@ -4,27 +4,22 @@ import com.fasterxml.jackson.databind.JsonNode; import com.fasterxml.jackson.databind.ObjectMapper; import shine.db.entities.UserSettingEntry; -import java.net.URI; -import java.net.http.HttpClient; -import java.net.http.WebSocket; -import java.time.Duration; import java.util.ArrayList; import java.util.List; -import java.util.UUID; -import java.util.concurrent.CompletableFuture; -import java.util.concurrent.CompletionStage; -import java.util.concurrent.CountDownLatch; -import java.util.concurrent.TimeUnit; -import java.util.concurrent.TimeoutException; public final class RemoteUserSettingsSyncClient { private static final ObjectMapper MAPPER = new ObjectMapper(); - private static final HttpClient HTTP = HttpClient.newBuilder() - .connectTimeout(Duration.ofSeconds(6)) - .build(); - public void upsertUserSetting(String serverAddressRaw, UserSettingEntry entry, boolean syncDelivery) throws Exception { - try (RemoteSyncSession session = new RemoteSyncSession(serverAddressRaw)) { + public void upsertUserSetting( + String serverAddressRaw, UserSettingEntry entry, boolean syncDelivery + ) throws Exception { + upsertUserSetting(null, serverAddressRaw, entry, syncDelivery); + } + + public void upsertUserSetting( + String targetServerLogin, String serverAddressRaw, UserSettingEntry entry, boolean syncDelivery + ) throws Exception { + try (RemoteSyncSession session = new RemoteSyncSession(targetServerLogin, serverAddressRaw)) { upsertUserSetting(session, entry, syncDelivery); } } @@ -69,7 +64,20 @@ public final class RemoteUserSettingsSyncClient { int limit, int maxBytes ) throws Exception { - try (RemoteSyncSession session = new RemoteSyncSession(serverAddressRaw)) { + return userSettingsSyncBatch(null, serverAddressRaw, ownerLogin, + afterTimeMs, afterSettingKey, limit, maxBytes); + } + + public RemoteUserSettingsBatch userSettingsSyncBatch( + String targetServerLogin, + String serverAddressRaw, + String ownerLogin, + long afterTimeMs, + String afterSettingKey, + int limit, + int maxBytes + ) throws Exception { + try (RemoteSyncSession session = new RemoteSyncSession(targetServerLogin, serverAddressRaw)) { return userSettingsSyncBatch(session, ownerLogin, afterTimeMs, afterSettingKey, limit, maxBytes); } } @@ -129,34 +137,6 @@ public final class RemoteUserSettingsSyncClient { ); } - private JsonNode send(String serverAddressRaw, String jsonTemplate) throws Exception { - String requestId = MAPPER.writeValueAsString("user-settings-sync-" + UUID.randomUUID()); - String json = jsonTemplate.formatted(requestId); - String wsUrl = RemoteBlockchainSyncClient.buildWsUrl(serverAddressRaw); - if (wsUrl == null) { - throw new IllegalArgumentException("Invalid server address: " + serverAddressRaw); - } - - CompletableFuture 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"); - } - - webSocket.sendText(json, true).get(8, TimeUnit.SECONDS); - String responseJson = responseFuture.get(12, TimeUnit.SECONDS); - tryAbort(webSocket); - return MAPPER.readTree(responseJson); - } - private void ensureOk(String op, JsonNode response) { int status = response.path("status").asInt(500); if (status >= 200 && status < 300) return; @@ -165,17 +145,6 @@ public final class RemoteUserSettingsSyncClient { throw new IllegalStateException(op + " failed: status=" + status + " code=" + code); } - private static void tryAbort(WebSocket webSocket) { - try { - webSocket.sendClose(WebSocket.NORMAL_CLOSURE, "ok"); - } catch (Exception ignored) { - } - try { - webSocket.abort(); - } catch (Exception ignored) { - } - } - public record RemoteUserSettingsBatch( long nextTimeMs, String nextSettingKey, @@ -194,35 +163,4 @@ public final class RemoteUserSettingsSyncClient { String signature ) {} - private static final class SyncWsListener implements WebSocket.Listener { - private final CompletableFuture responseFuture; - private final CountDownLatch openLatch; - private final StringBuilder textBuffer = new StringBuilder(); - - private SyncWsListener(CompletableFuture 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 void onError(WebSocket webSocket, Throwable error) { - responseFuture.completeExceptionally(error); - } - } } diff --git a/SHiNE-server/src/main/java/server/sync/BlockchainResyncRecoveryOnStartup.java b/SHiNE-server/src/main/java/server/sync/BlockchainResyncRecoveryOnStartup.java index 4d6755f0..8c79f513 100644 --- a/SHiNE-server/src/main/java/server/sync/BlockchainResyncRecoveryOnStartup.java +++ b/SHiNE-server/src/main/java/server/sync/BlockchainResyncRecoveryOnStartup.java @@ -67,7 +67,8 @@ public final class BlockchainResyncRecoveryOnStartup { blockchainName, partnerLogin, partnerAddress); try { - List heads = REMOTE.listBlockchainHeads(partnerAddress); + List heads = + REMOTE.listBlockchainHeads(partnerLogin, partnerAddress); RemoteBlockchainSyncClient.RemoteBlockchainHead remoteHead = heads.stream() .filter(h -> h != null && blockchainName.equals(h.blockchainName())) .findFirst() diff --git a/SHiNE-server/src/main/java/server/sync/PeriodicBlockchainSyncService.java b/SHiNE-server/src/main/java/server/sync/PeriodicBlockchainSyncService.java index 0b3deebe..a007d58d 100644 --- a/SHiNE-server/src/main/java/server/sync/PeriodicBlockchainSyncService.java +++ b/SHiNE-server/src/main/java/server/sync/PeriodicBlockchainSyncService.java @@ -102,7 +102,7 @@ public final class PeriodicBlockchainSyncService { if (partnerLogin == null) return; List remoteHeads = - REMOTE.listBlockchainHeads(partner.getServerAddress()); + REMOTE.listBlockchainHeads(partner.getLogin(), partner.getServerAddress()); for (RemoteBlockchainSyncClient.RemoteBlockchainHead remoteHead : remoteHeads) { if (remoteHead == null || remoteHead.blockchainName() == null || remoteHead.blockchainName().isBlank()) { @@ -170,7 +170,8 @@ public final class PeriodicBlockchainSyncService { int fromBlockNumber = Math.max(localLast + 1, 0); for (int blockNumber = fromBlockNumber; blockNumber <= remoteHead.lastBlockNumber(); blockNumber++) { RemoteBlockchainSyncClient.RemoteBlockchainBlock remoteBlock = - REMOTE.getBlockchainBlock(partner.getServerAddress(), remoteHead.blockchainName(), blockNumber); + REMOTE.getBlockchainBlock( + partner.getLogin(), partner.getServerAddress(), remoteHead.blockchainName(), blockNumber); if (remoteBlock == null) { log.warn("Periodic blockchain sync: remote block not found. partner={} blockchainName={} blockNumber={}", partnerLogin, remoteHead.blockchainName(), blockNumber); @@ -284,7 +285,8 @@ public final class PeriodicBlockchainSyncService { String localPrevHash = ""; for (int blockNumber = 0; blockNumber <= remoteHead.lastBlockNumber(); blockNumber++) { RemoteBlockchainSyncClient.RemoteBlockchainBlock remoteBlock = - REMOTE.getBlockchainBlock(partner.getServerAddress(), blockchainName, blockNumber); + REMOTE.getBlockchainBlock( + partner.getLogin(), partner.getServerAddress(), blockchainName, blockNumber); if (remoteBlock == null) { log.warn("Blockchain resync: remote block not found. partner={} blockchainName={} blockNumber={}", partnerLogin, blockchainName, blockNumber); diff --git a/SHiNE-server/src/main/java/server/sync/PeriodicUserSettingsSyncService.java b/SHiNE-server/src/main/java/server/sync/PeriodicUserSettingsSyncService.java index b8445b4a..b1fc6ad5 100644 --- a/SHiNE-server/src/main/java/server/sync/PeriodicUserSettingsSyncService.java +++ b/SHiNE-server/src/main/java/server/sync/PeriodicUserSettingsSyncService.java @@ -137,7 +137,7 @@ public final class PeriodicUserSettingsSyncService { boolean bootstrapCompleted = false; int appliedDm; - try (RemoteSyncSession session = new RemoteSyncSession(route.getServerUrl())) { + try (RemoteSyncSession session = new RemoteSyncSession(route.getServerLogin(), route.getServerUrl())) { for (int page = 0; page < maxPages; page++) { RemoteUserSettingsSyncClient.RemoteUserSettingsBatch batch = REMOTE.userSettingsSyncBatch( session, diff --git a/SHiNE-server/src/main/java/server/ws/README.md b/SHiNE-server/src/main/java/server/ws/README.md index 9d7a10e4..4ade637d 100644 --- a/SHiNE-server/src/main/java/server/ws/README.md +++ b/SHiNE-server/src/main/java/server/ws/README.md @@ -83,3 +83,11 @@ WebSocket-эндпоинт для одного соединения. - `WsServer` = сервер, который слушает порт и вешает `/ws`. - `BlockchainWsEndpoint` = обработчик одного WebSocket-подключения, мост между сетью и логикой. + +## Постоянные server-to-server соединения + +Исходящие межсерверные JSON-запросы проходят через +`server.sync.ServerConnectionPool`. Пул держит одно постоянное исходящее WSS- +соединение на `serverLogin`, отправляет `ServerHello`, поддерживает канал через +WebSocket ping/pong и переподключает его с backoff. DM, settings и blockchain +используют это соединение с разными приоритетами, не меняя свою бизнес-логику. diff --git a/SHiNE-server/src/main/java/server/ws/WsServer.java b/SHiNE-server/src/main/java/server/ws/WsServer.java index b504009a..03f095ac 100644 --- a/SHiNE-server/src/main/java/server/ws/WsServer.java +++ b/SHiNE-server/src/main/java/server/ws/WsServer.java @@ -12,6 +12,7 @@ import server.sync.PeriodicDmSyncService; import server.sync.PeriodicUserSettingsSyncService; import server.sync.SolanaUsersSyncStartupService; import server.sync.SyncServersBootstrapService; +import server.sync.ServerConnectionPool; import utils.config.AppConfig; import java.time.Duration; @@ -104,6 +105,10 @@ public final class WsServer { server.start(); log.info("✅ WS сервер запущен на ws://localhost:{}/ws", port); + ServerConnectionPool.getInstance().startOrLog(); + Runtime.getRuntime().addShutdownHook(new Thread( + () -> ServerConnectionPool.getInstance().close(), + "server-connection-pool-shutdown")); PeriodicDmSyncService.startOrLog(); PeriodicUserSettingsSyncService.startOrLog(); server.join(); diff --git a/SHiNE-server/src/main/resources/application.properties b/SHiNE-server/src/main/resources/application.properties index 374250c5..990103f7 100644 --- a/SHiNE-server/src/main/resources/application.properties +++ b/SHiNE-server/src/main/resources/application.properties @@ -41,6 +41,12 @@ dm.worker.dueLimit=100 dm.sync.batchLimit=200 dm.sync.batchMaxBytes=3000000 dm.sync.maxPagesPerPeer=20 +server.pool.pingIdleSeconds=120 +server.pool.pongTimeoutSeconds=15 +server.pool.requestTimeoutSeconds=12 +server.pool.connectTimeoutSeconds=15 +server.pool.callerTimeoutSeconds=35 +server.pool.maxQueuePerPeer=2000 server.info.url= server.info.physicalRegion= server.info.description= diff --git a/deploy/backup/backup-version.properties b/deploy/backup/backup-version.properties index 72dd5eb9..17bc0c72 100644 --- a/deploy/backup/backup-version.properties +++ b/deploy/backup/backup-version.properties @@ -1,3 +1,3 @@ backup.schema.version=2 -backup.full.version=3 -last.full.backup.date=2026-08-08 +backup.full.version=4 +last.full.backup.date=2026-08-25 diff --git a/docs/API/09_Operations_Index.md b/docs/API/09_Operations_Index.md index c50f48d0..db86f293 100644 --- a/docs/API/09_Operations_Index.md +++ b/docs/API/09_Operations_Index.md @@ -32,6 +32,7 @@ | `CloseActiveSession` | `03_Session_Management_API.md` | закрытие активной сессии | | `AddBlock` | `04_Add_Block_to_Blockchain_API.md` | добавление блока в блокчейн | | `GetBlockchainBlock` | `04_Add_Block_to_Blockchain_API.md` | чтение одного блока блокчейна | +| `ServerHello` | `16_Server_Connection_Pool_API.md` | объявление server-to-server соединения и возможностей peer без криптографической проверки | | `Ping` | `05_Technical_Requests_API.md` | keep-alive | | `GetServerInfo` | `05_Technical_Requests_API.md` | публичная информация о сервере | | `ListBlockchainHeads` | `05_Technical_Requests_API.md` | список heads всех локальных блокчейнов | @@ -79,6 +80,7 @@ - `ReceiveOutcomingMessage` зарегистрирован как алиас того же handler/request-класса, что и `SendMessagePair`, и сохраняет прежний межсерверный payload. - Межсерверные DM-операции пока доверяют `sourceServerLogin`; отдельная межсерверная авторизация запланирована позднее. +- `ServerHello` пока принимает заявленный `serverLogin` на доверии и не является криптографической аутентификацией. - Отдельных HTTP endpoints для DM-файлов сейчас нет. - Классы `Net_MarkChannelMessagesSeen_*` существуют в коде, но операция `MarkChannelMessagesSeen` не зарегистрирована в `JsonHandlerRegistry`, поэтому в публичный список API не входит. - HTTP debug endpoints из `src/main/java/server/debug/` не входят в этот индекс WebSocket `op`; они описаны отдельно в `13_HTTP_Debug_API.md`. diff --git a/docs/API/13_User_Settings_API.md b/docs/API/13_User_Settings_API.md index 3cecdacb..15135dac 100644 --- a/docs/API/13_User_Settings_API.md +++ b/docs/API/13_User_Settings_API.md @@ -83,7 +83,10 @@ ## 5. Синхронизация -Настройки и DM имеют раздельные таблицы и правила ACK, но периодический процесс открывает один последовательный WS-сеанс с peer: сначала синхронизирует настройки, затем забирает `DmSyncBatch`. +Настройки и DM имеют раздельные таблицы и правила ACK, но периодический процесс +использует один логический последовательный сеанс поверх постоянного WSS-пула: +сначала синхронизирует настройки, затем забирает `DmSyncBatch`. Завершение +логического сеанса не закрывает физический сокет. - локальная запись создаётся с `synced=false`, если её ещё не подтвердил второй сервер; - если запись пришла с другого сервера, она сохраняется сразу как `synced=true`; diff --git a/docs/Blockchain/sync-between-servers.md b/docs/Blockchain/sync-between-servers.md index 79fc7fc8..d951a25a 100644 --- a/docs/Blockchain/sync-between-servers.md +++ b/docs/Blockchain/sync-between-servers.md @@ -183,16 +183,20 @@ Full resync запускается только тогда, когда: Настройка влияет именно на этап подготовки отсутствующей локальной цепочки во время periodic sync. -## 5. Возможное развитие server-to-server транспорта +## 5. Реализованный постоянный server-to-server транспорт -Этот раздел не описывает текущую DM-доставку. DM уже использует короткие one-shot WebSocket-вызовы, indexed outbox и ACK. Ниже остаётся возможное развитие постоянного транспорта и server-auth. +Этот раздел не меняет текущую семантику DM, settings и blockchain. Он описывает +единый постоянный WSS-транспорт, через который выполняются уже существующие операции. ### 5.1 Межсерверное соединение -- Серверы устанавливают постоянное WebSocket-соединение друг с другом. +- Серверы устанавливают постоянное исходящее WebSocket-соединение друг с другом. - Адрес партнёра определяется по `server_address` из его Solana PDA. -- Аутентификация: подпись Ed25519 корневым ключом сервера (`root_key` из PDA). -- При разрыве — переподключение с экспоненциальным backoff. +- После подключения отправляется `ServerHello` с `serverLogin`, версией протокола и capabilities. +- На текущем этапе `serverLogin` принимается на доверии; подпись Ed25519 корневым ключом сервера отложена. +- При разрыве выполняется переподключение с jitter/backoff до 60 секунд. +- После 120 секунд отсутствия полезного трафика отправляется WebSocket ping; pong ожидается 15 секунд. +- Один физический канал переиспользуют DM, настройки и blockchain. ### 5.2 Доставка новых данных (push) @@ -246,19 +250,21 @@ Full resync запускается только тогда, когда: | Плановый blockchain sync при старте + каждые 12 часов | ✅ Реализовано | | Обход Solana RPC через `sync.importUserProfileFromPartner.enabled` | ✅ Реализовано | | Обычный `AddBlock` через `tmp_bch`/`write_check`/`write_pending` | ✅ Реализовано | -| Межсерверный постоянный WebSocket-канал | Нужна реализация | +| Межсерверный постоянный WebSocket-канал | ✅ Реализован общий `ServerConnectionPool` | | Асинхронная доставка DM на access-серверы получателя | ✅ Реализовано | | Retry DM до 1 часа + UI-state | ✅ Реализовано | | Репликация DM на второй access-сервер по `synced` | ✅ Реализовано | | Read-only `GetDmDeliveryStatus` | ✅ Реализовано | -| Push блоков блокчейна партнёрам | ✅ Реализована базовая one-shot версия | +| Push блоков блокчейна партнёрам | ✅ Выполняется через постоянный WSS-пул | | Periodic backfill отсутствующего хвоста | ✅ Реализовано | | Разрешение рассинхрона / divergence | ✅ Реализована базовая full-resync схема во время periodic sync | | Startup recovery по `*.resync_pending` marker-file | ✅ Реализовано | | Маршрутизация DM через один/два `access_servers` | ✅ Реализовано | | Криптографическая server-to-server авторизация DM | Нужна реализация | -Текущая версия сервера умеет синхронизацию блокчейнов и DM. Постоянные server-to-server соединения не требуются для текущей one-shot WS-реализации; отдельной будущей задачей остаётся криптографическая авторизация DM-вызовов. +Текущая версия сервера использует постоянный WSS-пул для существующих +server-to-server JSON-операций. Отдельной будущей задачей остаётся +криптографическая авторизация server-to-server вызовов. Следующие отдельные шаги после текущего этапа: - отдельно проверить full-resync и startup-recovery на реальном тестовом прогоне после ручного удаления БД/файлов.