Исправить определение клиентского DM-вызова

2026-08-25 19:30:44 +04:00
parent d8e0c77951
commit 7d929e626e
9 changed files with 996 additions and 6 deletions
@@ -0,0 +1,70 @@
+package server.logic.ws_protocol.JSON.handlers.system;
+
+import server.logic.ws_protocol.JSON.ConnectionContext;
+import server.logic.ws_protocol.JSON.entyties.Net_Request;
+import server.logic.ws_protocol.JSON.entyties.Net_Response;
+import server.logic.ws_protocol.JSON.handlers.JsonMessageHandler;
+import server.logic.ws_protocol.JSON.handlers.system.entyties.Net_ServerHello_Request;
+import server.logic.ws_protocol.JSON.handlers.system.entyties.Net_ServerHello_Response;
+import server.logic.ws_protocol.JSON.utils.NetExceptionResponseFactory;
+import server.logic.ws_protocol.WireCodes;
+import utils.config.AppConfig;
+
+import java.util.ArrayList;
+import java.util.LinkedHashSet;
+import java.util.List;
+import java.util.Locale;
+
+/**
+ * Временное server-to-server представление без криптографической проверки.
+ * Заявленный serverLogin принимается на доверии и привязывается к WS-контексту.
+ */
+public final class Net_ServerHello_Handler implements JsonMessageHandler {
+    private static final int PROTOCOL_VERSION = 1;
+    private static final List<String> CAPABILITIES = List.of(
+            "dm-sync", "settings-sync", "block-sync", "connection-pool");
+
+    @Override
+    public Net_Response handle(Net_Request baseRequest, ConnectionContext ctx) {
+        Net_ServerHello_Request req = (Net_ServerHello_Request) baseRequest;
+        String remoteLogin = normalize(req.getServerLogin());
+        int remoteVersion = req.getProtocolVersion() == null ? 0 : req.getProtocolVersion();
+        if (ctx == null) {
+            return NetExceptionResponseFactory.error(
+                    req, WireCodes.Status.BAD_REQUEST, "NO_CONNECTION_CONTEXT", "ServerHello требует WebSocket-контекст");
+        }
+        if (remoteLogin == null || remoteVersion <= 0) {
+            return NetExceptionResponseFactory.error(
+                    req, WireCodes.Status.BAD_REQUEST, "BAD_SERVER_HELLO", "serverLogin/protocolVersion обязательны");
+        }
+
+        LinkedHashSet<String> unique = new LinkedHashSet<>();
+        if (req.getCapabilities() != null) {
+            for (String capability : req.getCapabilities()) {
+                String normalized = normalize(capability);
+                if (normalized != null && unique.size() < 64) unique.add(normalized);
+            }
+        }
+        ctx.setServerConnection(true);
+        ctx.setRemoteServerLogin(remoteLogin);
+        ctx.setRemoteServerProtocolVersion(remoteVersion);
+        ctx.setRemoteServerCapabilities(new ArrayList<>(unique));
+
+        Net_ServerHello_Response resp = new Net_ServerHello_Response();
+        resp.setOp(req.getOp());
+        resp.setRequestId(req.getRequestId());
+        resp.setStatus(WireCodes.Status.OK);
+        resp.setAccepted(true);
+        String localLogin = normalize(AppConfig.getInstance().getParam("server.SHiNE.login"));
+        resp.setServerLogin(localLogin == null ? "" : localLogin);
+        resp.setProtocolVersion(PROTOCOL_VERSION);
+        resp.setCapabilities(CAPABILITIES);
+        return resp;
+    }
+
+    private static String normalize(String value) {
+        if (value == null) return null;
+        String normalized = value.trim().toLowerCase(Locale.ROOT);
+        return normalized.isEmpty() ? null : normalized;
+    }
+}
@@ -0,0 +1,20 @@
+package server.logic.ws_protocol.JSON.handlers.system.entyties;
+
+import server.logic.ws_protocol.JSON.entyties.Net_Request;
+
+import java.util.List;
+
+public final class Net_ServerHello_Request extends Net_Request {
+    private String serverLogin;
+    private Integer protocolVersion;
+    private List<String> capabilities;
+
+    public String getServerLogin() { return serverLogin; }
+    public void setServerLogin(String serverLogin) { this.serverLogin = serverLogin; }
+
+    public Integer getProtocolVersion() { return protocolVersion; }
+    public void setProtocolVersion(Integer protocolVersion) { this.protocolVersion = protocolVersion; }
+
+    public List<String> getCapabilities() { return capabilities; }
+    public void setCapabilities(List<String> capabilities) { this.capabilities = capabilities; }
+}
@@ -0,0 +1,24 @@
+package server.logic.ws_protocol.JSON.handlers.system.entyties;
+
+import server.logic.ws_protocol.JSON.entyties.Net_Response;
+
+import java.util.List;
+
+public final class Net_ServerHello_Response extends Net_Response {
+    private boolean accepted;
+    private String serverLogin;
+    private int protocolVersion;
+    private List<String> capabilities;
+
+    public boolean isAccepted() { return accepted; }
+    public void setAccepted(boolean accepted) { this.accepted = accepted; }
+
+    public String getServerLogin() { return serverLogin; }
+    public void setServerLogin(String serverLogin) { this.serverLogin = serverLogin; }
+
+    public int getProtocolVersion() { return protocolVersion; }
+    public void setProtocolVersion(int protocolVersion) { this.protocolVersion = protocolVersion; }
+
+    public List<String> getCapabilities() { return capabilities; }
+    public void setCapabilities(List<String> capabilities) { this.capabilities = capabilities; }
+}
@@ -48,8 +48,7 @@ public class Net_SendMessagePair_Handler implements JsonMessageHandler {

        SignedMessageEntry incomingEntry;
        SignedMessageEntry outgoingEntry;
-        boolean fromPeer = "ReceiveOutcomingMessage".equalsIgnoreCase(req.getOp())
-                || !isBlank(req.getSourceServerLogin());
+        boolean fromPeer = !isBlank(req.getSourceServerLogin());
        try {
            String sourceApi = fromPeer ? "ReceiveOutcomingMessage" : "SendMessagePair";
            String originSessionId = (ctx != null && !isBlank(ctx.getSessionId())) ? ctx.getSessionId() : null;
@@ -0,0 +1,763 @@
+package server.sync;
+
+import com.fasterxml.jackson.databind.JsonNode;
+import com.fasterxml.jackson.databind.ObjectMapper;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+import shine.db.dao.SyncServersDAO;
+import shine.db.dao.UserAccessServersCurrentDAO;
+import shine.db.entities.SyncServerEntry;
+import shine.db.entities.UserAccessServerRouteEntry;
+import utils.config.AppConfig;
+
+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.Comparator;
+import java.util.LinkedHashMap;
+import java.util.List;
+import java.util.Locale;
+import java.util.Map;
+import java.util.UUID;
+import java.util.concurrent.CompletableFuture;
+import java.util.concurrent.CompletionStage;
+import java.util.concurrent.ConcurrentHashMap;
+import java.util.concurrent.Executors;
+import java.util.concurrent.PriorityBlockingQueue;
+import java.util.concurrent.ScheduledExecutorService;
+import java.util.concurrent.ThreadFactory;
+import java.util.concurrent.ThreadPoolExecutor;
+import java.util.concurrent.TimeUnit;
+import java.util.concurrent.TimeoutException;
+import java.util.concurrent.atomic.AtomicBoolean;
+import java.util.concurrent.atomic.AtomicInteger;
+import java.util.concurrent.atomic.AtomicLong;
+
+/**
+ * Общий пул постоянных WSS-соединений между физическими серверами SHiNE.
+ *
+ * <p>Пул меняет только транспорт: существующие JSON-операции, ACK, outbox и
+ * расписания повторов остаются обязанностью вызывающих сервисов.</p>
+ */
+public final class ServerConnectionPool implements AutoCloseable {
+    private static final Logger log = LoggerFactory.getLogger(ServerConnectionPool.class);
+    private static final ObjectMapper MAPPER = new ObjectMapper();
+    private static final String SERVER_LOGIN_CONFIG = "server.SHiNE.login";
+    private static final int REQUEST_WORKERS = Math.max(
+            8, Math.min(32, Runtime.getRuntime().availableProcessors() * 2));
+    private static final int CONNECTION_WORKERS = Math.max(
+            4, Math.min(16, Runtime.getRuntime().availableProcessors()));
+    private static final ServerConnectionPool INSTANCE = new ServerConnectionPool();
+
+    private final HttpClient http = HttpClient.newBuilder()
+            .connectTimeout(Duration.ofSeconds(6))
+            .build();
+    private final ConcurrentHashMap<String, PeerConnection> peers = new ConcurrentHashMap<>();
+    private final AtomicBoolean started = new AtomicBoolean(false);
+    private final AtomicBoolean closed = new AtomicBoolean(false);
+    private final AtomicLong requestSequence = new AtomicLong();
+
+    private final ThreadPoolExecutor requestExecutor = new ThreadPoolExecutor(
+            REQUEST_WORKERS,
+            REQUEST_WORKERS,
+            60L,
+            TimeUnit.SECONDS,
+            new java.util.concurrent.LinkedBlockingQueue<>(10_000),
+            daemonThreadFactory("server-pool-request"),
+            new ThreadPoolExecutor.CallerRunsPolicy());
+    private final ThreadPoolExecutor connectionExecutor = new ThreadPoolExecutor(
+            CONNECTION_WORKERS,
+            CONNECTION_WORKERS,
+            60L,
+            TimeUnit.SECONDS,
+            new java.util.concurrent.LinkedBlockingQueue<>(10_000),
+            daemonThreadFactory("server-pool-connect"),
+            new ThreadPoolExecutor.DiscardPolicy());
+    private final ScheduledExecutorService scheduler = Executors.newScheduledThreadPool(
+            2, daemonThreadFactory("server-pool-scheduler"));
+
+    private ServerConnectionPool() {}
+
+    public static ServerConnectionPool getInstance() {
+        return INSTANCE;
+    }
+
+    public void startOrLog() {
+        if (closed.get() || !started.compareAndSet(false, true)) return;
+        scheduler.scheduleWithFixedDelay(this::refreshKnownPeersSafe, 0L, 30L, TimeUnit.SECONDS);
+        scheduler.scheduleWithFixedDelay(this::healthCheckSafe, 10L, 10L, TimeUnit.SECONDS);
+        scheduler.scheduleWithFixedDelay(this::logMetricsSafe, 5L, 5L, TimeUnit.MINUTES);
+        log.info("Server connection pool started: adaptive ping={}s pongTimeout={}s",
+                pingIdleSeconds(), pongTimeoutSeconds());
+    }
+
+    public JsonNode request(
+            String serverLogin,
+            String serverAddress,
+            String jsonTemplate,
+            Priority priority
+    ) throws Exception {
+        startOrLog();
+        String normalizedLogin = normalizeLogin(serverLogin);
+        if (normalizedLogin == null) {
+            normalizedLogin = loginFromAddress(serverAddress);
+        }
+        if (normalizedLogin == null) {
+            throw new IllegalArgumentException("Server login and address are empty");
+        }
+        PeerConnection peer = registerPeer(normalizedLogin, serverAddress);
+        return peer.request(jsonTemplate, priority == null ? Priority.NORMAL : priority);
+    }
+
+    public PeerConnection registerPeer(String serverLogin, String serverAddress) {
+        String login = normalizeLogin(serverLogin);
+        String wsUrl = buildWsUrl(serverAddress);
+        if (login == null || wsUrl == null) {
+            throw new IllegalArgumentException("Invalid server peer: login=" + serverLogin + " address=" + serverAddress);
+        }
+        PeerConnection peer = peers.computeIfAbsent(login, ignored -> new PeerConnection(login, wsUrl));
+        peer.updateWsUrl(wsUrl);
+        peer.ensureConnectedInBackground(0L);
+        return peer;
+    }
+
+    public List<PeerMetricsSnapshot> snapshotMetrics() {
+        List<PeerMetricsSnapshot> result = new ArrayList<>();
+        for (PeerConnection peer : peers.values()) result.add(peer.snapshot());
+        result.sort(Comparator.comparing(PeerMetricsSnapshot::serverLogin));
+        return List.copyOf(result);
+    }
+
+    private void refreshKnownPeersSafe() {
+        if (closed.get()) return;
+        try {
+            String ownLogin = normalizeLogin(AppConfig.getInstance().getParam(SERVER_LOGIN_CONFIG));
+            Map<String, String> discovered = new LinkedHashMap<>();
+            for (SyncServerEntry entry : SyncServersDAO.getInstance().listAll()) {
+                if (entry == null) continue;
+                putDiscovered(discovered, ownLogin, entry.getLogin(), entry.getServerAddress());
+            }
+            for (UserAccessServerRouteEntry entry : UserAccessServersCurrentDAO.getInstance().listDistinctServers()) {
+                if (entry == null) continue;
+                putDiscovered(discovered, ownLogin, entry.getServerLogin(), entry.getServerUrl());
+            }
+            for (Map.Entry<String, String> entry : discovered.entrySet()) {
+                try {
+                    registerPeer(entry.getKey(), entry.getValue());
+                } catch (Exception e) {
+                    log.warn("Server pool peer registration failed: server={} reason={}",
+                            entry.getKey(), compactError(e));
+                }
+            }
+        } catch (Exception e) {
+            log.warn("Server pool peer refresh failed: {}", compactError(e));
+        }
+    }
+
+    private static void putDiscovered(
+            Map<String, String> discovered,
+            String ownLogin,
+            String serverLogin,
+            String serverAddress
+    ) {
+        String login = normalizeLogin(serverLogin);
+        if (login == null || login.equals(ownLogin) || buildWsUrl(serverAddress) == null) return;
+        discovered.put(login, serverAddress);
+    }
+
+    private void healthCheckSafe() {
+        if (closed.get()) return;
+        for (PeerConnection peer : peers.values()) {
+            try {
+                peer.healthCheck();
+            } catch (Exception e) {
+                log.debug("Server pool health check failed: server={} reason={}",
+                        peer.serverLogin, compactError(e));
+            }
+        }
+    }
+
+    private void logMetricsSafe() {
+        if (closed.get() || peers.isEmpty()) return;
+        long connected = peers.values().stream().filter(p -> p.state == ConnectionState.CONNECTED).count();
+        int queued = peers.values().stream().mapToInt(p -> p.queuedCount.get()).sum();
+        long errors = peers.values().stream().mapToLong(p -> p.failedRequests.get()).sum();
+        log.info("Server pool metrics: peers={} connected={} queued={} failedRequests={}",
+                peers.size(), connected, queued, errors);
+    }
+
+    @Override
+    public void close() {
+        if (!closed.compareAndSet(false, true)) return;
+        for (PeerConnection peer : peers.values()) peer.closeConnection("pool_shutdown");
+        scheduler.shutdownNow();
+        requestExecutor.shutdownNow();
+        connectionExecutor.shutdownNow();
+        peers.clear();
+    }
+
+    public enum Priority {
+        REALTIME(0),
+        NORMAL(1),
+        BULK(2);
+
+        private final int rank;
+        Priority(int rank) { this.rank = rank; }
+    }
+
+    public enum ConnectionState {
+        DISCONNECTED,
+        CONNECTING,
+        CONNECTED,
+        CLOSED
+    }
+
+    public record PeerMetricsSnapshot(
+            String serverLogin,
+            String wsUrl,
+            ConnectionState state,
+            long connectedAtMs,
+            long lastActivityAtMs,
+            long lastPingAtMs,
+            long lastPongAtMs,
+            long reconnectCount,
+            int queuedRealtime,
+            int queuedNormal,
+            int queuedBulk,
+            long successfulRequests,
+            long failedRequests,
+            long timedOutRequests,
+            String lastError
+    ) {}
+
+    public final class PeerConnection {
+        private final String serverLogin;
+        private final PriorityBlockingQueue<QueuedRequest> queue = new PriorityBlockingQueue<>();
+        private final ConcurrentHashMap<String, CompletableFuture<JsonNode>> pending = new ConcurrentHashMap<>();
+        private final AtomicBoolean drainScheduled = new AtomicBoolean(false);
+        private final AtomicBoolean reconnectScheduled = new AtomicBoolean(false);
+        private final AtomicInteger queuedCount = new AtomicInteger();
+        private final AtomicInteger reconnectAttempt = new AtomicInteger();
+        private final AtomicLong generation = new AtomicLong();
+        private final AtomicLong reconnectCount = new AtomicLong();
+        private final AtomicLong successfulRequests = new AtomicLong();
+        private final AtomicLong failedRequests = new AtomicLong();
+        private final AtomicLong timedOutRequests = new AtomicLong();
+        private final Object connectLock = new Object();
+
+        private volatile String wsUrl;
+        private volatile ConnectionState state = ConnectionState.DISCONNECTED;
+        private volatile WebSocket webSocket;
+        private volatile CompletableFuture<Void> readiness;
+        private volatile long connectedAtMs;
+        private volatile long lastActivityAtMs = System.currentTimeMillis();
+        private volatile long lastPingAtMs;
+        private volatile long lastPongAtMs;
+        private volatile long pingAwaitedSinceMs;
+        private volatile String lastError = "";
+
+        private PeerConnection(String serverLogin, String wsUrl) {
+            this.serverLogin = serverLogin;
+            this.wsUrl = wsUrl;
+        }
+
+        private JsonNode request(String jsonTemplate, Priority priority) throws Exception {
+            if (jsonTemplate == null || jsonTemplate.isBlank()) {
+                throw new IllegalArgumentException("JSON request template is empty");
+            }
+            int maxQueue = (int) configLong("server.pool.maxQueuePerPeer", 2_000L, 10L, 100_000L);
+            int queued = queuedCount.incrementAndGet();
+            if (queued > maxQueue) {
+                queuedCount.decrementAndGet();
+                throw new IllegalStateException("Server peer queue is full: " + serverLogin);
+            }
+
+            QueuedRequest task = new QueuedRequest(
+                    priority,
+                    requestSequence.incrementAndGet(),
+                    jsonTemplate,
+                    new CompletableFuture<>());
+            queue.offer(task);
+            scheduleDrain();
+
+            long timeoutSeconds = configLong("server.pool.callerTimeoutSeconds", 35L, 5L, 120L);
+            try {
+                return task.result.get(timeoutSeconds, TimeUnit.SECONDS);
+            } catch (java.util.concurrent.TimeoutException e) {
+                timedOutRequests.incrementAndGet();
+                if (queue.remove(task)) queuedCount.decrementAndGet();
+                throw new TimeoutException("Server pool caller timeout: " + serverLogin);
+            }
+        }
+
+        private void scheduleDrain() {
+            if (!drainScheduled.compareAndSet(false, true)) return;
+            requestExecutor.execute(this::drainQueue);
+        }
+
+        private void drainQueue() {
+            try {
+                for (;;) {
+                    QueuedRequest task = queue.poll();
+                    if (task == null) return;
+                    queuedCount.decrementAndGet();
+                    if (task.result.isDone()) continue;
+                    try {
+                        ensureConnectedBlocking();
+                        JsonNode response = sendTemplate(task.jsonTemplate)
+                                .get(requestTimeoutSeconds(), TimeUnit.SECONDS);
+                        successfulRequests.incrementAndGet();
+                        task.result.complete(response);
+                    } catch (java.util.concurrent.TimeoutException e) {
+                        timedOutRequests.incrementAndGet();
+                        failedRequests.incrementAndGet();
+                        task.result.completeExceptionally(
+                                new TimeoutException("Server pool response timeout: " + serverLogin));
+                        invalidateConnection("response_timeout", e);
+                    } catch (Exception e) {
+                        failedRequests.incrementAndGet();
+                        task.result.completeExceptionally(unwrap(e));
+                        if (state != ConnectionState.DISCONNECTED) {
+                            invalidateConnection("request_failed", e);
+                        }
+                    }
+                }
+            } finally {
+                drainScheduled.set(false);
+                if (!queue.isEmpty()) scheduleDrain();
+            }
+        }
+
+        private void ensureConnectedBlocking() throws Exception {
+            CompletableFuture<Void> localReady;
+            synchronized (connectLock) {
+                if (state == ConnectionState.CONNECTED && webSocket != null && !webSocket.isOutputClosed()) return;
+                if (state == ConnectionState.CLOSED || closed.get()) {
+                    throw new IllegalStateException("Server connection pool is closed");
+                }
+                if (readiness == null || readiness.isDone()) startConnectLocked();
+                localReady = readiness;
+            }
+            localReady.get(connectAndHelloTimeoutSeconds(), TimeUnit.SECONDS);
+        }
+
+        private void startConnectLocked() {
+            long currentGeneration = generation.incrementAndGet();
+            state = ConnectionState.CONNECTING;
+            lastError = "";
+            CompletableFuture<Void> ready = new CompletableFuture<>();
+            readiness = ready;
+            Listener listener = new Listener(this, currentGeneration);
+
+            http.newWebSocketBuilder()
+                    .connectTimeout(Duration.ofSeconds(6))
+                    .buildAsync(URI.create(wsUrl), listener)
+                    .thenCompose(ws -> {
+                        if (generation.get() != currentGeneration) {
+                            try { ws.abort(); } catch (Exception ignored) {}
+                            return CompletableFuture.failedFuture(new IllegalStateException("Superseded server connection"));
+                        }
+                        webSocket = ws;
+                        return sendServerHello(ws);
+                    })
+                    .whenComplete((ignored, error) -> {
+                        if (error != null) {
+                            ready.completeExceptionally(unwrap(error));
+                            invalidateConnection(currentGeneration, "connect_or_hello_failed", error);
+                            return;
+                        }
+                        if (generation.get() != currentGeneration) {
+                            ready.completeExceptionally(new IllegalStateException("Superseded server connection"));
+                            return;
+                        }
+                        long now = System.currentTimeMillis();
+                        connectedAtMs = now;
+                        lastActivityAtMs = now;
+                        lastPongAtMs = now;
+                        pingAwaitedSinceMs = 0L;
+                        reconnectAttempt.set(0);
+                        state = ConnectionState.CONNECTED;
+                        ready.complete(null);
+                        log.info("Server pool connected: server={} url={}", serverLogin, wsUrl);
+                    });
+        }
+
+        private CompletableFuture<Void> sendServerHello(WebSocket ws) {
+            try {
+                String ownLogin = normalizeLogin(AppConfig.getInstance().getParam(SERVER_LOGIN_CONFIG));
+                if (ownLogin == null) ownLogin = "unconfigured-server";
+                String requestId = "server-hello-" + UUID.randomUUID();
+                CompletableFuture<JsonNode> response = new CompletableFuture<>();
+                pending.put(requestId, response);
+                String json = """
+                    {
+                      "op":"ServerHello",
+                      "requestId":%s,
+                      "payload":{
+                        "serverLogin":%s,
+                        "protocolVersion":1,
+                        "capabilities":["dm-sync","settings-sync","block-sync","connection-pool"]
+                      }
+                    }
+                    """.formatted(
+                        MAPPER.writeValueAsString(requestId),
+                        MAPPER.writeValueAsString(ownLogin));
+                lastActivityAtMs = System.currentTimeMillis();
+                ws.sendText(json, true).whenComplete((ignored, error) -> {
+                    if (error != null) {
+                        pending.remove(requestId);
+                        response.completeExceptionally(error);
+                    }
+                });
+                return response.orTimeout(requestTimeoutSeconds(), TimeUnit.SECONDS)
+                        .thenAccept(node -> ensureOk("ServerHello", node));
+            } catch (Exception e) {
+                return CompletableFuture.failedFuture(e);
+            }
+        }
+
+        private CompletableFuture<JsonNode> sendTemplate(String jsonTemplate) throws Exception {
+            WebSocket ws = webSocket;
+            if (state != ConnectionState.CONNECTED || ws == null || ws.isOutputClosed()) {
+                return CompletableFuture.failedFuture(new IllegalStateException("Server peer is disconnected"));
+            }
+            String requestId = "server-pool-" + UUID.randomUUID();
+            String requestIdJson = MAPPER.writeValueAsString(requestId);
+            String json = fillRequestId(jsonTemplate, requestIdJson);
+            CompletableFuture<JsonNode> response = new CompletableFuture<>();
+            pending.put(requestId, response);
+            lastActivityAtMs = System.currentTimeMillis();
+            ws.sendText(json, true).whenComplete((ignored, error) -> {
+                if (error != null) {
+                    pending.remove(requestId);
+                    response.completeExceptionally(error);
+                }
+            });
+            return response;
+        }
+
+        private void acceptText(long listenerGeneration, String text) {
+            if (generation.get() != listenerGeneration || text == null || text.isBlank()) return;
+            lastActivityAtMs = System.currentTimeMillis();
+            try {
+                JsonNode node = MAPPER.readTree(text);
+                String requestId = node.path("requestId").asText("");
+                CompletableFuture<JsonNode> response = requestId.isBlank() ? null : pending.remove(requestId);
+                if (response != null) {
+                    response.complete(node);
+                } else {
+                    log.debug("Server pool ignored unmatched frame: server={} requestId={} op={}",
+                            serverLogin, requestId, node.path("op").asText(""));
+                }
+            } catch (Exception e) {
+                log.warn("Server pool received invalid JSON: server={} reason={}", serverLogin, compactError(e));
+            }
+        }
+
+        private void healthCheck() {
+            if (state == ConnectionState.DISCONNECTED) {
+                ensureConnectedInBackground(reconnectDelayMillis(reconnectAttempt.get()));
+                return;
+            }
+            if (state != ConnectionState.CONNECTED) return;
+            long now = System.currentTimeMillis();
+            if (pingAwaitedSinceMs > 0L) {
+                if (now - pingAwaitedSinceMs >= TimeUnit.SECONDS.toMillis(pongTimeoutSeconds())) {
+                    invalidateConnection("pong_timeout", new TimeoutException("Pong timeout"));
+                }
+                return;
+            }
+            if (now - lastActivityAtMs < TimeUnit.SECONDS.toMillis(pingIdleSeconds())) return;
+            WebSocket ws = webSocket;
+            if (ws == null || ws.isOutputClosed()) {
+                invalidateConnection("socket_closed", null);
+                return;
+            }
+            long nonce = now;
+            pingAwaitedSinceMs = now;
+            lastPingAtMs = now;
+            ws.sendPing(ByteBuffer.allocate(Long.BYTES).putLong(0, nonce))
+                    .whenComplete((ignored, error) -> {
+                        if (error != null) invalidateConnection("ping_failed", error);
+                    });
+        }
+
+        private void acceptPong(long listenerGeneration) {
+            if (generation.get() != listenerGeneration) return;
+            long now = System.currentTimeMillis();
+            lastPongAtMs = now;
+            lastActivityAtMs = now;
+            pingAwaitedSinceMs = 0L;
+        }
+
+        private void updateWsUrl(String newWsUrl) {
+            if (newWsUrl.equals(wsUrl)) return;
+            wsUrl = newWsUrl;
+            invalidateConnection("peer_url_changed", null);
+        }
+
+        private void ensureConnectedInBackground(long delayMs) {
+            if (closed.get() || state == ConnectionState.CLOSED) return;
+            if (!reconnectScheduled.compareAndSet(false, true)) return;
+            scheduler.schedule(() -> {
+                reconnectScheduled.set(false);
+                if (closed.get() || state == ConnectionState.CLOSED || state == ConnectionState.CONNECTED) return;
+                connectionExecutor.execute(() -> {
+                    try {
+                        ensureConnectedBlocking();
+                    } catch (Exception e) {
+                        lastError = compactError(e);
+                        // Ошибка connect/ServerHello сама инвалидирует поколение и планирует reconnect.
+                        // Здесь второй schedule дал бы двойной рост backoff для одной попытки.
+                    }
+                });
+            }, Math.max(0L, delayMs), TimeUnit.MILLISECONDS);
+        }
+
+        private void scheduleReconnect() {
+            if (closed.get() || state == ConnectionState.CLOSED) return;
+            int attempt = reconnectAttempt.getAndUpdate(value -> Math.min(value + 1, 30));
+            reconnectCount.incrementAndGet();
+            ensureConnectedInBackground(reconnectDelayMillis(attempt));
+        }
+
+        private long reconnectDelayMillis(int attempt) {
+            long[] seconds = {1L, 2L, 4L, 8L, 15L, 30L, 60L};
+            long base = seconds[Math.min(Math.max(0, attempt), seconds.length - 1)];
+            long halfMs = TimeUnit.SECONDS.toMillis(base) / 2L;
+            long jitterMs = java.util.concurrent.ThreadLocalRandom.current().nextLong(halfMs + 1L);
+            return halfMs + jitterMs;
+        }
+
+        private void invalidateConnection(String reason, Throwable error) {
+            invalidateConnection(generation.get(), reason, error);
+        }
+
+        private void invalidateConnection(long expectedGeneration, String reason, Throwable error) {
+            if (state == ConnectionState.CLOSED
+                    || !generation.compareAndSet(expectedGeneration, expectedGeneration + 1L)) return;
+            lastError = error == null ? reason : reason + ": " + compactError(error);
+            state = ConnectionState.DISCONNECTED;
+            pingAwaitedSinceMs = 0L;
+            WebSocket ws = webSocket;
+            webSocket = null;
+            if (ws != null) {
+                try { ws.abort(); } catch (Exception ignored) {}
+            }
+            Exception failure = new IllegalStateException("Server connection lost: " + serverLogin + " (" + reason + ")");
+            for (Map.Entry<String, CompletableFuture<JsonNode>> entry : pending.entrySet()) {
+                if (pending.remove(entry.getKey(), entry.getValue())) {
+                    entry.getValue().completeExceptionally(failure);
+                }
+            }
+            scheduleReconnect();
+        }
+
+        private void closeConnection(String reason) {
+            state = ConnectionState.CLOSED;
+            generation.incrementAndGet();
+            WebSocket ws = webSocket;
+            webSocket = null;
+            if (ws != null) {
+                try { ws.sendClose(WebSocket.NORMAL_CLOSURE, reason); } catch (Exception ignored) {}
+                try { ws.abort(); } catch (Exception ignored) {}
+            }
+            IllegalStateException failure = new IllegalStateException("Server connection closed: " + serverLogin);
+            for (CompletableFuture<JsonNode> future : pending.values()) future.completeExceptionally(failure);
+            pending.clear();
+            for (QueuedRequest request : queue) request.result.completeExceptionally(failure);
+            queue.clear();
+            queuedCount.set(0);
+        }
+
+        private PeerMetricsSnapshot snapshot() {
+            int realtime = 0;
+            int normal = 0;
+            int bulk = 0;
+            for (QueuedRequest request : queue) {
+                switch (request.priority) {
+                    case REALTIME -> realtime++;
+                    case NORMAL -> normal++;
+                    case BULK -> bulk++;
+                }
+            }
+            return new PeerMetricsSnapshot(
+                    serverLogin, wsUrl, state, connectedAtMs, lastActivityAtMs,
+                    lastPingAtMs, lastPongAtMs, reconnectCount.get(), realtime, normal, bulk,
+                    successfulRequests.get(), failedRequests.get(), timedOutRequests.get(), lastError);
+        }
+    }
+
+    private record QueuedRequest(
+            Priority priority,
+            long sequence,
+            String jsonTemplate,
+            CompletableFuture<JsonNode> result
+    ) implements Comparable<QueuedRequest> {
+        @Override
+        public int compareTo(QueuedRequest other) {
+            int byPriority = Integer.compare(priority.rank, other.priority.rank);
+            return byPriority != 0 ? byPriority : Long.compare(sequence, other.sequence);
+        }
+    }
+
+    private static final class Listener implements WebSocket.Listener {
+        private final PeerConnection peer;
+        private final long generation;
+        private final StringBuilder text = new StringBuilder();
+
+        private Listener(PeerConnection peer, long generation) {
+            this.peer = peer;
+            this.generation = generation;
+        }
+
+        @Override
+        public void onOpen(WebSocket webSocket) {
+            webSocket.request(1);
+        }
+
+        @Override
+        public CompletionStage<?> onText(WebSocket webSocket, CharSequence data, boolean last) {
+            text.append(data);
+            if (last) {
+                peer.acceptText(generation, text.toString());
+                text.setLength(0);
+            }
+            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<?> onPong(WebSocket webSocket, ByteBuffer message) {
+            peer.acceptPong(generation);
+            webSocket.request(1);
+            return CompletableFuture.completedFuture(null);
+        }
+
+        @Override
+        public CompletionStage<?> onClose(WebSocket webSocket, int statusCode, String reason) {
+            peer.invalidateConnection(generation, "remote_close_" + statusCode, null);
+            return CompletableFuture.completedFuture(null);
+        }
+
+        @Override
+        public void onError(WebSocket webSocket, Throwable error) {
+            peer.invalidateConnection(generation, "websocket_error", error);
+        }
+    }
+
+    public static String buildWsUrl(String serverAddressRaw) {
+        if (serverAddressRaw == null) return null;
+        String raw = serverAddressRaw.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;
+            int port = uri.getPort();
+            String authority = host.trim().toLowerCase(Locale.ROOT) + (port > 0 ? ":" + port : "");
+            return "wss://" + authority + "/ws";
+        } catch (Exception e) {
+            return null;
+        }
+    }
+
+    private static String loginFromAddress(String serverAddress) {
+        String wsUrl = buildWsUrl(serverAddress);
+        if (wsUrl == null) return null;
+        try {
+            return "address:" + URI.create(wsUrl).getAuthority().toLowerCase(Locale.ROOT);
+        } catch (Exception e) {
+            return null;
+        }
+    }
+
+    private static String normalizeLogin(String value) {
+        if (value == null) return null;
+        String normalized = value.trim().toLowerCase(Locale.ROOT);
+        return normalized.isEmpty() ? null : normalized;
+    }
+
+    private static void ensureOk(String op, JsonNode response) {
+        int status = response == null ? 500 : response.path("status").asInt(500);
+        if (status >= 200 && status < 300) return;
+        String code = response == null ? "EMPTY_RESPONSE" : response.path("code").asText("");
+        if (code.isBlank() && response != null) code = response.path("error").asText("");
+        throw new IllegalStateException(op + " failed: status=" + status + " code=" + code);
+    }
+
+    private static String fillRequestId(String template, String requestIdJson) {
+        int marker = template.indexOf("%s");
+        if (marker < 0) throw new IllegalArgumentException("JSON template has no requestId marker");
+        return template.substring(0, marker) + requestIdJson + template.substring(marker + 2);
+    }
+
+    private static Throwable unwrap(Throwable error) {
+        Throwable current = error;
+        while ((current instanceof java.util.concurrent.CompletionException
+                || current instanceof java.util.concurrent.ExecutionException)
+                && current.getCause() != null) {
+            current = current.getCause();
+        }
+        return current;
+    }
+
+    private static Exception unwrap(Exception error) {
+        Throwable unwrapped = unwrap((Throwable) error);
+        return unwrapped instanceof Exception e ? e : new IllegalStateException(unwrapped);
+    }
+
+    private static String compactError(Throwable error) {
+        String text = String.valueOf(error == null ? "unknown" : error.getMessage());
+        return text.length() <= 500 ? text : text.substring(0, 500);
+    }
+
+    private static long requestTimeoutSeconds() {
+        return configLong("server.pool.requestTimeoutSeconds", 12L, 3L, 120L);
+    }
+
+    private static long connectAndHelloTimeoutSeconds() {
+        return configLong("server.pool.connectTimeoutSeconds", 15L, 5L, 120L);
+    }
+
+    private static long pingIdleSeconds() {
+        return configLong("server.pool.pingIdleSeconds", 120L, 15L, 240L);
+    }
+
+    private static long pongTimeoutSeconds() {
+        return configLong("server.pool.pongTimeoutSeconds", 15L, 5L, 120L);
+    }
+
+    private static long configLong(String key, long defaultValue, long min, long max) {
+        String raw = AppConfig.getInstance().getParam(key);
+        if (raw == null || raw.isBlank()) return defaultValue;
+        try {
+            long value = Long.parseLong(raw.trim());
+            return Math.max(min, Math.min(max, value));
+        } catch (Exception ignored) {
+            return defaultValue;
+        }
+    }
+
+    private static ThreadFactory daemonThreadFactory(String prefix) {
+        return new ThreadFactory() {
+            private final AtomicLong sequence = new AtomicLong();
+            @Override
+            public Thread newThread(Runnable r) {
+                Thread thread = new Thread(r, prefix + "-" + sequence.incrementAndGet());
+                thread.setDaemon(true);
+                return thread;
+            }
+        };
+    }
+}
@@ -152,7 +152,7 @@
 }
 ```

-`sourceServerLogin` необязателен для совместимости. Пока межсерверная авторизация отложена, это поле считается доверенным. Пользовательская подпись signed-блока проверяется всегда.
+`sourceServerLogin` необязателен для совместимости. Пока межсерверная авторизация отложена, это поле считается доверенным. Пользовательская подпись signed-блока проверяется всегда. Пустой `sourceServerLogin` трактуется как клиентский вызов, непустой - как peer-вызов.

 Успешный ответ содержит существующие `messageKey`, `baseKey` и счётчики realtime-доставки. Повтор уже сохранённой той же ревизии обрабатывается идемпотентно.

@@ -356,7 +356,11 @@ Pull-синхронизация событий владельца с `synced=fal

 Ответ содержит `messageKey`, `known` и `delivered`. `delivered=true` означает доставку хотя бы одному серверу получателя.

-Периодический процесс использует одно WS-соединение с peer: сначала синхронизирует настройки, затем вызывает `DmSyncBatch` до завершения страниц и ACK. Существующий `MarkAllUserSettingsUnsynced` также сбрасывает DM-флаги и возвращает `dmUpdated`; отдельной операции `MarkAllDmUnsynced` нет.
+Периодический процесс использует один логический последовательный сеанс поверх
+постоянного WSS-пула: сначала синхронизирует настройки, затем вызывает
+`DmSyncBatch` до завершения страниц и ACK. Существующий
+`MarkAllUserSettingsUnsynced` также сбрасывает DM-флаги и возвращает
+`dmUpdated`; отдельной операции `MarkAllDmUnsynced` нет.

 Основные ошибки межсерверных операций:

@@ -0,0 +1,105 @@
+# Межсерверное соединение и `ServerHello`
+
+Документ описывает транспортный слой постоянных WSS-соединений между серверами SHiNE.
+Он не меняет форматы DM, блоков, настроек, ACK или расписания повторных попыток.
+
+## 1. `ServerHello`
+
+После установления исходящего WSS-соединения сервер первым запросом отправляет:
+
+```json
+{
+  "op": "ServerHello",
+  "requestId": "server-hello-001",
+  "payload": {
+    "serverLogin": "shineupme",
+    "protocolVersion": 1,
+    "capabilities": [
+      "dm-sync",
+      "settings-sync",
+      "block-sync",
+      "connection-pool"
+    ]
+  }
+}
+```
+
+Успешный ответ:
+
+```json
+{
+  "op": "ServerHello",
+  "requestId": "server-hello-001",
+  "status": 200,
+  "payload": {
+    "accepted": true,
+    "serverLogin": "server2",
+    "protocolVersion": 1,
+    "capabilities": [
+      "dm-sync",
+      "settings-sync",
+      "block-sync",
+      "connection-pool"
+    ]
+  }
+}
+```
+
+На текущем этапе `serverLogin` принимается на доверии. Подпись, challenge и
+проверка корневого ключа сервера намеренно отложены. Поэтому `ServerHello`
+фиксирует тип соединения и возможности peer, но пока не является
+криптографической аутентификацией.
+
+## 2. Пул соединений
+
+- физическое соединение создаётся одно на `serverLogin`;
+- логические операции DM, settings и blockchain используют один WSS;
+- завершение `RemoteSyncSession` не закрывает физический сокет;
+- известные peer берутся из `sync_servers` и `user_access_servers_current`;
+- список перечитывается каждые 30 секунд;
+- при изменении URL соединение пересоздаётся;
+- новые запросы не повторяются транспортом автоматически: действующие
+  domain-воркеры сохраняют прежние правила retry и идемпотентности.
+
+## 3. Приоритеты
+
+| Приоритет | Операции |
+| --- | --- |
+| `REALTIME` | доставка DM, deletes, `GetDmDeliveryStatus` |
+| `NORMAL` | настройки, `DmSyncBatch` и access-data sync |
+| `BULK` | blockchain heads, blocks, `AddBlock` backfill |
+
+На одном peer одновременно исполняется один запрос. Приоритет применяется к
+ожидающей очереди и не прерывает уже начатую операцию.
+
+## 4. Ping, reconnect и таймауты
+
+Значения по умолчанию:
+
+| Параметр | Значение |
+| --- | ---: |
+| `server.pool.pingIdleSeconds` | `120` |
+| `server.pool.pongTimeoutSeconds` | `15` |
+| `server.pool.requestTimeoutSeconds` | `12` |
+| `server.pool.connectTimeoutSeconds` | `15` |
+| `server.pool.callerTimeoutSeconds` | `35` |
+| `server.pool.maxQueuePerPeer` | `2000` |
+
+Ping отправляется WebSocket control-frame только после периода отсутствия
+полезного трафика. При потере соединения применяется reconnect с jitter и
+ступенями до 60 секунд.
+
+## 5. Диагностика
+
+Внутренний snapshot пула хранит для каждого peer:
+
+- состояние соединения;
+- URL;
+- времена connect/activity/ping/pong;
+- число reconnect;
+- размеры очередей по приоритетам;
+- успешные, ошибочные и просроченные запросы;
+- последнюю ошибку.
+
+Агрегированное состояние периодически записывается в серверный лог. Отдельная
+публичная операция метрик на этом этапе не добавляется.
@@ -103,6 +103,8 @@ DM считается доставленным, когда signed-входящу

 При обрыве соединения неподтверждённый элемент остаётся `synced=false` и безопасно приходит повторно. Элемент, полученный от peer, локально сразу считается синхронизированным, чтобы не образовалась петля.

+`sourceServerLogin` является единственным признаком межсерверного вызова для этих операций: если поле пустое, запрос считается клиентским и его запись не должна сразу переводиться в `synced=true`.
+
 Синхронизация настроек и DM проходит последовательно через один WS-сеанс: сначала настройки, затем все страницы DM. Если второй сервер был выключен, после включения он сам догружает пропущенные элементы.

 Существующий `MarkAllUserSettingsUnsynced` также сбрасывает DM-флаги. После сброса история повторно передаётся как синхронизация, но старые сообщения получателю заново не отправляются: delivery-состояние создаётся с учётом их возраста и не открывает завершённую часовую очередь.
@@ -489,7 +489,7 @@ Request:
 }
 ```

-`sourceServerLogin` пока доверяется без отдельной межсерверной подписи. Сервер всё равно проверяет пользовательскую подпись самого `SHiNE_DM`. Успешный ответ сохраняет старые поля `messageKey`, `baseKey` и счётчики realtime-доставки.
+`sourceServerLogin` пока доверяется без отдельной межсерверной подписи. Сервер всё равно проверяет пользовательскую подпись самого `SHiNE_DM`. Для `ReceiveOutcomingMessage` и `SendMessagePair` пустой `sourceServerLogin` означает клиентский вызов, непустой - peer-вызов. Успешный ответ сохраняет старые поля `messageKey`, `baseKey` и счётчики realtime-доставки.

 ### 10.3. `DeleteMessage`

@@ -616,7 +616,10 @@ UI-следствие для клиента:
 - не отправляет realtime/push-уведомления клиентам;
 - не запускает повторный fan-out, чтобы не создавать циклы.

-Синхронизация настроек и DM выполняется одним периодическим процессом и через один последовательный WS-сеанс с peer. Выборка DM использует частичный индекс по `synced=false`.
+Синхронизация настроек и DM выполняется одним периодическим процессом и через
+один логический последовательный сеанс поверх постоянного WSS-пула. Закрытие
+этого логического сеанса не закрывает физическое соединение с peer. Выборка DM
+использует частичный индекс по `synced=false`.

 В текущей реализации межсерверная авторизация DM ещё не включена. Принимающий сервер проверяет, что сам является access-сервером `ownerLogin`, и всегда проверяет пользовательские подписи signed-блоков.