From ef389cc2374b86e6ffea217e4779c3ef7ecc92a8 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?=D0=93=D1=80=D0=B8=D0=B3=D0=BE=D1=80=D0=B8=D0=B9=20=D0=A1?= =?UTF-8?q?=D0=B0=D1=84=D1=80=D0=BE=D0=BD=D0=BE=D0=B2?= Date: Thu, 11 Jun 2026 17:36:47 +0000 Subject: [PATCH] Delete internal/cluster/node.go --- internal/cluster/node.go | 974 --------------------------------------- 1 file changed, 974 deletions(-) delete mode 100644 internal/cluster/node.go diff --git a/internal/cluster/node.go b/internal/cluster/node.go deleted file mode 100644 index 6a4e2c9..0000000 --- a/internal/cluster/node.go +++ /dev/null @@ -1,974 +0,0 @@ -/* - * Copyright 2026 Safronov Grigorii - * - * Licensed under the CDDL, Version 1.0 (the "License"); - * you may not use this file except in compliance with the License. - * - * You may obtain a copy of the License at - * https://opensource.org/licenses/CDDL-1.0 - */ - -// Файл: internal/cluster/node.go -// Назначение: Реализация узла кластера (node) для распределённой СУБД с поддержкой временных меток. -// Полностью lock-free с использованием атомарных операций. - -package cluster - -import ( - "crypto/rand" - "encoding/base64" - "encoding/json" - "fmt" - "io" - "net" - "runtime/debug" - "sync" - "sync/atomic" - "time" - - "futriis/internal/log" - "futriis/internal/storage" - "github.com/google/uuid" -) - -// NodeStatus представляет состояние узла кластера -type NodeStatus int32 - -const ( - StatusOffline NodeStatus = iota - StatusActive - StatusSyncing - StatusFailed -) - -// Node представляет отдельный узел в распределённой системе -type Node struct { - ID string // Уникальный идентификатор узла - IP string // IP-адрес узла - Port int // Порт для коммуникации - Status atomic.Int32 // Атомарный статус узла (NodeStatus) - Storage *storage.Storage - logger *log.Logger - coordinator *RaftCoordinator // Ссылка на координатора (теперь RaftCoordinator) - lastSeen atomic.Int64 // Время последнего heartbeat (Unix миллисекунды) - joinedAt atomic.Int64 // Время присоединения к кластеру - createdAt int64 // Время создания узла - startedAt int64 // Время последнего старта узла - stoppedAt int64 // Время остановки узла - incomingConn chan net.Conn // Канал для входящих соединений (wait-free) - stopChan chan struct{} - requestCount atomic.Uint64 // Счётчик обработанных запросов - bytesRx atomic.Uint64 // Получено байт - bytesTx atomic.Uint64 // Отправлено байт - mu sync.RWMutex // Для синхронизации изменений статуса - - // Новые компоненты для production-ready реализации - workerPool *WorkerPool // Пул воркеров для ограничения горутин - replicator *NetworkReplicator // Сетевой репликатор с retry и backoff - connPool sync.Map // Пул соединений к другим узлам -} - -// NodeConfig содержит конфигурацию для создания узла -type NodeConfig struct { - IP string - Port int - Storage *storage.Storage - Logger *log.Logger - Coordinator *RaftCoordinator -} - -// NewNode создаёт новый экземпляр узла кластера -func NewNode(ip string, port int, store *storage.Storage, logger *log.Logger) *Node { - now := time.Now().UnixMilli() - - // Создаём пул воркеров с ограничением (максимум 500 одновременных горутин) - workerPool := NewWorkerPool(500, logger) - - // Создаём сетевой репликатор с настройками retry и backoff - replicator := NewNetworkReplicator(DefaultReplicationRetryConfig(), workerPool, logger) - - node := &Node{ - ID: uuid.New().String(), - IP: ip, - Port: port, - Storage: store, - logger: logger, - incomingConn: make(chan net.Conn, 10000), // Увеличенный буфер - stopChan: make(chan struct{}), - createdAt: now, - startedAt: now, - workerPool: workerPool, - replicator: replicator, - } - node.Status.Store(int32(StatusActive)) - node.lastSeen.Store(now) - node.joinedAt.Store(0) - - // Безопасный запуск сервера с обработкой паник - SafeGoWithLogger(node.startTCPServer, logger, "TCPServer") - SafeGoWithLogger(node.handleIncomingConnections, logger, "IncomingConnections") - SafeGoWithLogger(node.heartbeatLoop, logger, "HeartbeatLoop") - SafeGoWithLogger(node.connectionHealthMonitor, logger, "ConnectionHealthMonitor") - - if logger != nil { - logger.Info(fmt.Sprintf("Node %s created at %s with worker pool (max: 500)", node.ID, node.GetCreatedAtStr())) - } - - return node -} - -// startTCPServer запускает TCP-сервер для приёма запросов от других узлов -func (n *Node) startTCPServer() { - defer func() { - if r := recover(); r != nil { - if n.logger != nil { - n.logger.Error(fmt.Sprintf("TCP server panicked: %v\n%s", r, debug.Stack())) - } - // Перезапускаем сервер - time.Sleep(5 * time.Second) - SafeGoWithLogger(n.startTCPServer, n.logger, "TCPServer") - } - }() - - addr := fmt.Sprintf("%s:%d", n.IP, n.Port) - listener, err := net.Listen("tcp", addr) - if err != nil { - if n.logger != nil { - n.logger.Error(fmt.Sprintf("Node %s failed to start TCP server: %v", n.ID, err)) - } - n.Status.Store(int32(StatusFailed)) - return - } - defer listener.Close() - - if n.logger != nil { - n.logger.Info(fmt.Sprintf("Node %s listening on %s (started at %s)", n.ID, addr, n.GetStartedAtStr())) - } - - for { - select { - case <-n.stopChan: - if n.logger != nil { - n.logger.Info(fmt.Sprintf("Node %s TCP server stopped", n.ID)) - } - return - default: - // Устанавливаем таймаут для Accept - if err := listener.(*net.TCPListener).SetDeadline(time.Now().Add(5 * time.Second)); err != nil { - continue - } - - conn, err := listener.Accept() - if err != nil { - if netErr, ok := err.(net.Error); ok && netErr.Timeout() { - continue - } - if n.logger != nil { - n.logger.Error(fmt.Sprintf("Node %s accept error: %v", n.ID, err)) - } - continue - } - - // Устанавливаем таймауты на соединение - conn.SetReadDeadline(time.Now().Add(30 * time.Second)) - conn.SetWriteDeadline(time.Now().Add(30 * time.Second)) - - select { - case n.incomingConn <- conn: - n.bytesRx.Add(1) - default: - if n.logger != nil { - n.logger.Warn(fmt.Sprintf("Node %s incoming connection queue full, dropping connection", n.ID)) - } - conn.Close() - } - } - } -} - -// handleIncomingConnections обрабатывает входящие соединения с использованием пула воркеров -func (n *Node) handleIncomingConnections() { - defer func() { - if r := recover(); r != nil { - if n.logger != nil { - n.logger.Error(fmt.Sprintf("Incoming connections handler panicked: %v\n%s", r, debug.Stack())) - } - // Перезапускаем обработчик - SafeGoWithLogger(n.handleIncomingConnections, n.logger, "IncomingConnections") - } - }() - - for { - select { - case <-n.stopChan: - return - case conn := <-n.incomingConn: - n.requestCount.Add(1) - // Используем пул воркеров вместо создания неограниченного количества горутин - taskID := fmt.Sprintf("handle_conn_%d_%s", time.Now().UnixNano(), conn.RemoteAddr().String()) - err := n.workerPool.SubmitFunc(taskID, func() error { - n.handleNodeRequest(conn) - return nil - }) - if err != nil { - if n.logger != nil { - n.logger.Warn(fmt.Sprintf("Failed to submit connection task: %v", err)) - } - conn.Close() - } - } - } -} - -// handleNodeRequest обрабатывает конкретный запрос от другого узла -func (n *Node) handleNodeRequest(conn net.Conn) { - defer func() { - if r := recover(); r != nil { - if n.logger != nil { - n.logger.Error(fmt.Sprintf("Request handler panicked: %v\n%s", r, debug.Stack())) - } - } - conn.Close() - }() - - // Устанавливаем таймаут на чтение - conn.SetReadDeadline(time.Now().Add(30 * time.Second)) - - decoder := json.NewDecoder(conn) - var req NodeRequest - if err := decoder.Decode(&req); err != nil { - if err != io.EOF && n.logger != nil { - n.logger.Error(fmt.Sprintf("Node %s failed to decode request: %v", n.ID, err)) - } - return - } - - n.lastSeen.Store(time.Now().UnixMilli()) - - if n.logger != nil { - n.logger.Debug(fmt.Sprintf("Node %s received request type %s from %s at %s", - n.ID, req.Type, req.FromNode, time.UnixMilli(req.Timestamp).Format("15:04:05.000"))) - } - - switch req.Type { - case "replicate": - n.handleReplicateRequest(req.Data) - case "query": - n.handleQueryRequest(req.Data, conn) - case "sync": - n.handleSyncRequest(req.Data, conn) - case "heartbeat": - n.handleHeartbeatRequest(req, conn) - case "status_sync": - n.handleStatusSyncRequest(req, conn) - default: - if n.logger != nil { - n.logger.Warn(fmt.Sprintf("Node %s unknown request type: %s", n.ID, req.Type)) - } - } -} - -// handleReplicateRequest обрабатывает запрос на репликацию документа -func (n *Node) handleReplicateRequest(data []byte) { - defer func() { - if r := recover(); r != nil { - if n.logger != nil { - n.logger.Error(fmt.Sprintf("Replicate request handler panicked: %v\n%s", r, debug.Stack())) - } - } - }() - - startTime := time.Now().UnixMilli() - - var repData struct { - Database string `json:"database"` - Collection string `json:"collection"` - Document map[string]interface{} `json:"document"` - SourceNode string `json:"source_node"` - ReplicaID string `json:"replica_id"` - } - - if err := json.Unmarshal(data, &repData); err != nil { - if n.logger != nil { - n.logger.Error(fmt.Sprintf("Node %s failed to unmarshal replicate data: %v", n.ID, err)) - } - return - } - - db, err := n.Storage.GetDatabase(repData.Database) - if err != nil { - if n.logger != nil { - n.logger.Error(fmt.Sprintf("Node %s database not found for replication: %s", n.ID, repData.Database)) - } - return - } - - coll, err := db.GetCollection(repData.Collection) - if err != nil { - if n.logger != nil { - n.logger.Error(fmt.Sprintf("Node %s collection not found for replication: %s", n.ID, repData.Collection)) - } - return - } - - // Безопасное получение ID документа - docID, ok := repData.Document["_id"].(string) - if !ok { - if n.logger != nil { - n.logger.Error(fmt.Sprintf("Node %s document missing _id field", n.ID)) - } - return - } - - doc := &storage.Document{ - ID: docID, - Fields: repData.Document, - } - - if err := coll.Insert(doc); err != nil { - if n.logger != nil { - n.logger.Error(fmt.Sprintf("Node %s failed to replicate document: %v", n.ID, err)) - } - } else { - duration := time.Now().UnixMilli() - startTime - if n.logger != nil { - n.logger.Debug(fmt.Sprintf("Node %s replicated document %s from %s (took %d ms)", - n.ID, doc.ID, repData.SourceNode, duration)) - } - } -} - -// handleQueryRequest обрабатывает запрос на чтение данных с узла -func (n *Node) handleQueryRequest(data []byte, conn net.Conn) { - defer func() { - if r := recover(); r != nil { - if n.logger != nil { - n.logger.Error(fmt.Sprintf("Query request handler panicked: %v\n%s", r, debug.Stack())) - } - n.sendErrorResponse(conn, "Internal server error") - } - }() - - startTime := time.Now().UnixMilli() - - var queryData struct { - Database string `json:"database"` - Collection string `json:"collection"` - DocumentID string `json:"document_id"` - RequestID string `json:"request_id"` - } - - if err := json.Unmarshal(data, &queryData); err != nil { - n.sendErrorResponse(conn, err.Error()) - return - } - - db, err := n.Storage.GetDatabase(queryData.Database) - if err != nil { - n.sendErrorResponse(conn, err.Error()) - return - } - - coll, err := db.GetCollection(queryData.Collection) - if err != nil { - n.sendErrorResponse(conn, err.Error()) - return - } - - doc, err := coll.Find(queryData.DocumentID) - if err != nil { - n.sendErrorResponse(conn, err.Error()) - return - } - - duration := time.Now().UnixMilli() - startTime - - response := map[string]interface{}{ - "status": "success", - "data": doc, - "node_id": n.ID, - "request_id": queryData.RequestID, - "duration_ms": duration, - "timestamp": time.Now().UnixMilli(), - } - - conn.SetWriteDeadline(time.Now().Add(10 * time.Second)) - encoder := json.NewEncoder(conn) - if err := encoder.Encode(response); err == nil { - responseData, _ := json.Marshal(response) - n.bytesTx.Add(uint64(len(responseData))) - } - - if n.logger != nil { - n.logger.Debug(fmt.Sprintf("Node %s handled query for %s.%s:%s (took %d ms)", - n.ID, queryData.Database, queryData.Collection, queryData.DocumentID, duration)) - } -} - -// handleSyncRequest обрабатывает запрос на синхронизацию всей коллекции -func (n *Node) handleSyncRequest(data []byte, conn net.Conn) { - defer func() { - if r := recover(); r != nil { - if n.logger != nil { - n.logger.Error(fmt.Sprintf("Sync request handler panicked: %v\n%s", r, debug.Stack())) - } - n.sendErrorResponse(conn, "Internal server error") - } - }() - - startTime := time.Now().UnixMilli() - - var syncData struct { - Database string `json:"database"` - Collection string `json:"collection"` - RequestID string `json:"request_id"` - Since int64 `json:"since"` - } - - if err := json.Unmarshal(data, &syncData); err != nil { - n.sendErrorResponse(conn, err.Error()) - return - } - - db, err := n.Storage.GetDatabase(syncData.Database) - if err != nil { - n.sendErrorResponse(conn, err.Error()) - return - } - - coll, err := db.GetCollection(syncData.Collection) - if err != nil { - n.sendErrorResponse(conn, err.Error()) - return - } - - docs := coll.GetAllDocuments() - - if syncData.Since > 0 { - filtered := make([]*storage.Document, 0) - for _, doc := range docs { - if doc.UpdatedAt > syncData.Since { - filtered = append(filtered, doc) - } - } - docs = filtered - } - - duration := time.Now().UnixMilli() - startTime - - response := map[string]interface{}{ - "status": "success", - "docs": docs, - "count": len(docs), - "node_id": n.ID, - "request_id": syncData.RequestID, - "duration_ms": duration, - "timestamp": time.Now().UnixMilli(), - "sync_duration_ms": duration, - } - - conn.SetWriteDeadline(time.Now().Add(30 * time.Second)) - encoder := json.NewEncoder(conn) - if err := encoder.Encode(response); err != nil && n.logger != nil { - n.logger.Error(fmt.Sprintf("Failed to send sync response: %v", err)) - } - - if n.logger != nil { - n.logger.Info(fmt.Sprintf("Node %s synced %d documents from %s.%s (took %d ms)", - n.ID, len(docs), syncData.Database, syncData.Collection, duration)) - } -} - -// handleHeartbeatRequest обрабатывает heartbeat запрос -func (n *Node) handleHeartbeatRequest(req NodeRequest, conn net.Conn) { - n.lastSeen.Store(time.Now().UnixMilli()) - - response := map[string]interface{}{ - "status": "alive", - "node_id": n.ID, - "timestamp": time.Now().UnixMilli(), - "uptime_ms": time.Now().UnixMilli() - n.startedAt, - } - - conn.SetWriteDeadline(time.Now().Add(5 * time.Second)) - encoder := json.NewEncoder(conn) - encoder.Encode(response) -} - -// handleStatusSyncRequest обрабатывает запрос синхронизации статуса (для split-brain recovery) -func (n *Node) handleStatusSyncRequest(req NodeRequest, conn net.Conn) { - var syncStatus struct { - LeaderID string `json:"leader_id"` - Term uint64 `json:"term"` - ClusterSize int `json:"cluster_size"` - } - - if err := json.Unmarshal(req.Data, &syncStatus); err != nil { - n.sendErrorResponse(conn, err.Error()) - return - } - - if n.coordinator != nil { - n.coordinator.HandleStatusSync(syncStatus.LeaderID, syncStatus.Term, syncStatus.ClusterSize) - } - - response := map[string]interface{}{ - "status": "synced", - "node_id": n.ID, - "term": n.coordinator.GetCurrentTerm(), - "is_leader": n.coordinator.IsLeader(), - "timestamp": time.Now().UnixMilli(), - } - - conn.SetWriteDeadline(time.Now().Add(10 * time.Second)) - encoder := json.NewEncoder(conn) - encoder.Encode(response) -} - -// sendErrorResponse отправляет ошибку в ответ на запрос -func (n *Node) sendErrorResponse(conn net.Conn, errMsg string) { - response := map[string]interface{}{ - "status": "error", - "error": errMsg, - "node_id": n.ID, - "timestamp": time.Now().UnixMilli(), - } - - conn.SetWriteDeadline(time.Now().Add(5 * time.Second)) - encoder := json.NewEncoder(conn) - encoder.Encode(response) -} - -// heartbeatLoop отправляет периодические сигналы жизни координатору -func (n *Node) heartbeatLoop() { - defer func() { - if r := recover(); r != nil { - if n.logger != nil { - n.logger.Error(fmt.Sprintf("Heartbeat loop panicked: %v\n%s", r, debug.Stack())) - } - // Перезапускаем heartbeat loop - time.Sleep(5 * time.Second) - SafeGoWithLogger(n.heartbeatLoop, n.logger, "HeartbeatLoop") - } - }() - - ticker := time.NewTicker(5 * time.Second) - defer ticker.Stop() - - for { - select { - case <-n.stopChan: - return - case <-ticker.C: - if n.coordinator != nil { - n.coordinator.SendHeartbeat(n.ID) - n.lastSeen.Store(time.Now().UnixMilli()) - if n.logger != nil { - n.logger.Debug(fmt.Sprintf("Node %s sent heartbeat at %s", n.ID, n.GetLastSeenStr())) - } - } - } - } -} - -// connectionHealthMonitor мониторит здоровье соединений -func (n *Node) connectionHealthMonitor() { - defer func() { - if r := recover(); r != nil { - if n.logger != nil { - n.logger.Error(fmt.Sprintf("Connection health monitor panicked: %v\n%s", r, debug.Stack())) - } - SafeGoWithLogger(n.connectionHealthMonitor, n.logger, "ConnectionHealthMonitor") - } - }() - - ticker := time.NewTicker(30 * time.Second) - defer ticker.Stop() - - for { - select { - case <-n.stopChan: - return - case <-ticker.C: - n.cleanupStaleConnections() - } - } -} - -// cleanupStaleConnections очищает старые соединения -func (n *Node) cleanupStaleConnections() { - n.connPool.Range(func(key, value interface{}) bool { - if conn, ok := value.(net.Conn); ok { - // Проверяем, активно ли соединение - conn.SetReadDeadline(time.Now().Add(1 * time.Second)) - buf := make([]byte, 1) - _, err := conn.Read(buf) - if err != nil { - conn.Close() - n.connPool.Delete(key) - if n.logger != nil { - n.logger.Debug(fmt.Sprintf("Cleaned up stale connection for %v", key)) - } - } - } - return true - }) -} - -// GetNodeStatus возвращает текущий статус узла (атомарно) -func (n *Node) GetNodeStatus() NodeStatus { - return NodeStatus(n.Status.Load()) -} - -// IsActive проверяет, активен ли узел -func (n *Node) IsActive() bool { - return NodeStatus(n.Status.Load()) == StatusActive -} - -// SetStatus устанавливает статус узла с синхронизацией через Raft -func (n *Node) SetStatus(status NodeStatus) error { - n.mu.Lock() - defer n.mu.Unlock() - - oldStatus := n.Status.Load() - if oldStatus == int32(status) { - return nil - } - - if n.coordinator != nil && n.coordinator.IsLeader() { - if err := n.coordinator.UpdateNodeStatus(n.ID, status); err != nil { - if n.logger != nil { - n.logger.Error(fmt.Sprintf("Failed to update node status via Raft: %v", err)) - } - return err - } - } - - n.Status.Store(int32(status)) - - if n.logger != nil { - n.logger.Info(fmt.Sprintf("Node %s status changed from %d to %d at %s", - n.ID, oldStatus, status, time.Now().Format("2006-01-02 15:04:05.000"))) - } - - return nil -} - -// SetCoordinator устанавливает координатора для узла -func (n *Node) SetCoordinator(coord *RaftCoordinator) { - n.coordinator = coord - now := time.Now().UnixMilli() - n.joinedAt.Store(now) - - if n.logger != nil { - n.logger.Info(fmt.Sprintf("Node %s joined cluster at %s", n.ID, n.GetJoinedAtStr())) - } -} - -// JoinCluster присоединяет узел к кластеру -func (n *Node) JoinCluster(coord *RaftCoordinator) error { - if n.coordinator != nil { - return fmt.Errorf("node already joined to cluster") - } - - n.SetCoordinator(coord) - - if err := coord.RegisterNode(n); err != nil { - return fmt.Errorf("failed to register node: %v", err) - } - - if err := n.SetStatus(StatusActive); err != nil { - return fmt.Errorf("failed to set active status: %v", err) - } - - if n.logger != nil { - n.logger.Info(fmt.Sprintf("Node %s successfully joined cluster at %s", n.ID, n.GetJoinedAtStr())) - } - - return nil -} - -// LeaveCluster покидает кластер -func (n *Node) LeaveCluster() error { - if n.coordinator == nil { - return fmt.Errorf("node not in cluster") - } - - if err := n.SetStatus(StatusOffline); err != nil { - n.logger.Warn(fmt.Sprintf("Failed to set offline status: %v", err)) - } - - if err := n.coordinator.RemoveNode(n.ID); err != nil { - n.logger.Warn(fmt.Sprintf("Failed to remove node from coordinator: %v", err)) - } - - n.coordinator = nil - n.joinedAt.Store(0) - - if n.logger != nil { - n.logger.Info(fmt.Sprintf("Node %s left cluster at %s", n.ID, time.Now().Format("2006-01-02 15:04:05.000"))) - } - - return nil -} - -// GetLastSeen возвращает время последнего контакта -func (n *Node) GetLastSeen() int64 { - return n.lastSeen.Load() -} - -// GetLastSeenStr возвращает человекочитаемое время последнего контакта -func (n *Node) GetLastSeenStr() string { - lastSeen := n.lastSeen.Load() - if lastSeen == 0 { - return "never" - } - return time.UnixMilli(lastSeen).Format("2006-01-02 15:04:05.000") -} - -// GetJoinedAt возвращает время присоединения к кластеру -func (n *Node) GetJoinedAt() int64 { - return n.joinedAt.Load() -} - -// GetJoinedAtStr возвращает человекочитаемое время присоединения -func (n *Node) GetJoinedAtStr() string { - joinedAt := n.joinedAt.Load() - if joinedAt == 0 { - return "not joined" - } - return time.UnixMilli(joinedAt).Format("2006-01-02 15:04:05.000") -} - -// GetStartedAt возвращает время старта узла -func (n *Node) GetStartedAt() int64 { - return n.startedAt -} - -// GetStartedAtStr возвращает человекочитаемое время старта -func (n *Node) GetStartedAtStr() string { - return time.UnixMilli(n.startedAt).Format("2006-01-02 15:04:05.000") -} - -// GetCreatedAt возвращает время создания узла -func (n *Node) GetCreatedAt() int64 { - return n.createdAt -} - -// GetCreatedAtStr возвращает человекочитаемое время создания -func (n *Node) GetCreatedAtStr() string { - return time.UnixMilli(n.createdAt).Format("2006-01-02 15:04:05.000") -} - -// GetUptime возвращает время работы узла -func (n *Node) GetUptime() time.Duration { - if n.startedAt == 0 { - return 0 - } - return time.Duration(time.Now().UnixMilli()-n.startedAt) * time.Millisecond -} - -// GetStats возвращает статистику узла -func (n *Node) GetStats() map[string]interface{} { - stats := map[string]interface{}{ - "id": n.ID, - "ip": n.IP, - "port": n.Port, - "status": n.GetNodeStatus(), - "created_at": n.GetCreatedAtStr(), - "started_at": n.GetStartedAtStr(), - "joined_at": n.GetJoinedAtStr(), - "last_seen": n.GetLastSeenStr(), - "uptime": n.GetUptime().String(), - "request_count": n.requestCount.Load(), - "bytes_rx": n.bytesRx.Load(), - "bytes_tx": n.bytesTx.Load(), - } - - if n.workerPool != nil { - stats["worker_pool"] = n.workerPool.GetStats() - } - - if n.replicator != nil { - stats["replication"] = n.replicator.GetStats() - } - - return stats -} - -// Stop останавливает работу узла -func (n *Node) Stop() { - if n.coordinator != nil && n.coordinator.IsLeader() { - n.SetStatus(StatusOffline) - } - n.Status.Store(int32(StatusOffline)) - n.stoppedAt = time.Now().UnixMilli() - - close(n.stopChan) - - if n.workerPool != nil { - n.workerPool.Stop() - } - - if n.replicator != nil { - n.replicator.Close() - } - - // Закрываем все соединения в пуле - n.connPool.Range(func(key, value interface{}) bool { - if conn, ok := value.(net.Conn); ok { - conn.Close() - } - return true - }) - - if n.logger != nil { - n.logger.Info(fmt.Sprintf("Node %s stopped at %s", n.ID, time.UnixMilli(n.stoppedAt).Format("2006-01-02 15:04:05.000"))) - } -} - -// GetAddress возвращает адрес узла в формате "ip:port" -func (n *Node) GetAddress() string { - return fmt.Sprintf("%s:%d", n.IP, n.Port) -} - -// generateReplicationID генерирует уникальный ID для репликации -func generateReplicationID() string { - bytes := make([]byte, 16) - rand.Read(bytes) - return base64.URLEncoding.EncodeToString(bytes) -} - -// ReplicateDocument отправляет документ на репликацию всем активным узлам с retry и backoff -func (n *Node) ReplicateDocument(database, collection string, doc *storage.Document) error { - if n.coordinator == nil { - if n.logger != nil { - n.logger.Warn("No coordinator set, skipping replication") - } - return fmt.Errorf("no coordinator set") - } - - nodes := n.coordinator.GetActiveNodes() - if len(nodes) <= 1 { - if n.logger != nil { - n.logger.Debug("No other nodes for replication") - } - return nil - } - - // Подготавливаем данные для репликации - repData := struct { - Database string `json:"database"` - Collection string `json:"collection"` - Document map[string]interface{} `json:"document"` - SourceNode string `json:"source_node"` - ReplicaID string `json:"replica_id"` - }{ - Database: database, - Collection: collection, - Document: doc.GetFields(), - SourceNode: n.ID, - ReplicaID: generateReplicationID(), - } - - data, err := json.Marshal(repData) - if err != nil { - return fmt.Errorf("failed to marshal replication data: %v", err) - } - - startTime := time.Now().UnixMilli() - - // Отправляем на все узлы, кроме себя - var wg sync.WaitGroup - var failedCount atomic.Int32 - var successCount atomic.Int32 - - for _, nodeInfo := range nodes { - if nodeInfo.ID == n.ID { - continue - } - - wg.Add(1) - targetNodeID := nodeInfo.ID - targetAddress := fmt.Sprintf("%s:%d", nodeInfo.IP, nodeInfo.Port) - - // Асинхронная репликация с повторными попытками через пул воркеров - taskID := fmt.Sprintf("replicate_%s_to_%s_%s", doc.ID, targetNodeID, repData.ReplicaID) - err := n.workerPool.SubmitFunc(taskID, func() error { - defer wg.Done() - - if n.replicator == nil { - failedCount.Add(1) - return fmt.Errorf("replicator not initialized") - } - - err := n.replicator.Replicate(targetNodeID, targetAddress, data) - if err != nil { - failedCount.Add(1) - if n.logger != nil { - n.logger.Error(fmt.Sprintf("Failed to replicate document %s to node %s after retries: %v", - doc.ID, targetNodeID, err)) - } - return err - } - - successCount.Add(1) - if n.logger != nil { - n.logger.Debug(fmt.Sprintf("Successfully replicated document %s to node %s", doc.ID, targetNodeID)) - } - return nil - }) - - if err != nil { - wg.Done() - failedCount.Add(1) - if n.logger != nil { - n.logger.Error(fmt.Sprintf("Failed to submit replication task for %s to %s: %v", - doc.ID, targetNodeID, err)) - } - } - } - - // Ждём завершения всех репликаций с таймаутом - done := make(chan struct{}) - go func() { - wg.Wait() - close(done) - }() - - select { - case <-done: - duration := time.Now().UnixMilli() - startTime - if n.logger != nil { - n.logger.Info(fmt.Sprintf("Replicated document %s to %d/%d nodes (took %d ms)", - doc.ID, successCount.Load(), len(nodes)-1, duration)) - } - case <-time.After(30 * time.Second): - if n.logger != nil { - n.logger.Warn(fmt.Sprintf("Replication timeout for document %s after %d ms", doc.ID, time.Now().UnixMilli()-startTime)) - } - return fmt.Errorf("replication timeout") - } - - if failedCount.Load() > 0 { - return fmt.Errorf("replication partially failed: %d of %d nodes failed", failedCount.Load(), len(nodes)-1) - } - - return nil -} - -// GetWorkerPoolStats возвращает статистику пула воркеров -func (n *Node) GetWorkerPoolStats() map[string]interface{} { - if n.workerPool == nil { - return map[string]interface{}{"enabled": false} - } - return n.workerPool.GetStats() -} - -// GetReplicationStats возвращает статистику репликации -func (n *Node) GetReplicationStats() map[string]interface{} { - if n.replicator == nil { - return map[string]interface{}{"enabled": false} - } - return n.replicator.GetStats() -}