430 lines
14 KiB
Go
430 lines
14 KiB
Go
/*
|
||
* 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/network_replication.go
|
||
// Назначение: Реальная сетевая репликация с повторными попытками и бэкоффом
|
||
|
||
package cluster
|
||
|
||
import (
|
||
"crypto/rand"
|
||
"encoding/binary"
|
||
"encoding/json"
|
||
"fmt"
|
||
"io"
|
||
"math"
|
||
"net"
|
||
"sync"
|
||
"sync/atomic"
|
||
"time"
|
||
)
|
||
|
||
// ReplicationRetryConfig содержит конфигурацию повторных попыток
|
||
type ReplicationRetryConfig struct {
|
||
MaxRetries int
|
||
InitialBackoff time.Duration
|
||
MaxBackoff time.Duration
|
||
BackoffFactor float64
|
||
JitterEnabled bool
|
||
}
|
||
|
||
// NetworkReplicationStats содержит статистику репликации
|
||
type NetworkReplicationStats struct {
|
||
TotalRequests atomic.Uint64
|
||
SuccessfulReqs atomic.Uint64
|
||
FailedReqs atomic.Uint64
|
||
RetriedReqs atomic.Uint64
|
||
AvgLatencyMs atomic.Uint64
|
||
BytesSent atomic.Uint64
|
||
BytesReceived atomic.Uint64
|
||
}
|
||
|
||
// DefaultReplicationRetryConfig возвращает конфигурацию по умолчанию
|
||
func DefaultReplicationRetryConfig() *ReplicationRetryConfig {
|
||
return &ReplicationRetryConfig{
|
||
MaxRetries: 5,
|
||
InitialBackoff: 100 * time.Millisecond,
|
||
MaxBackoff: 10 * time.Second,
|
||
BackoffFactor: 2.0,
|
||
JitterEnabled: true,
|
||
}
|
||
}
|
||
|
||
// validateConfig проверяет конфигурацию
|
||
func validateConfig(c *ReplicationRetryConfig) error {
|
||
if c.MaxRetries < 0 {
|
||
return fmt.Errorf("max_retries cannot be negative")
|
||
}
|
||
if c.InitialBackoff < 0 {
|
||
return fmt.Errorf("initial_backoff cannot be negative")
|
||
}
|
||
if c.MaxBackoff < c.InitialBackoff {
|
||
return fmt.Errorf("max_backoff must be >= initial_backoff")
|
||
}
|
||
if c.BackoffFactor < 1.0 {
|
||
return fmt.Errorf("backoff_factor must be >= 1.0")
|
||
}
|
||
return nil
|
||
}
|
||
|
||
// ReplicatedConnection представляет соединение с узлом
|
||
type ReplicatedConnection struct {
|
||
NodeID string
|
||
Address string
|
||
Conn net.Conn
|
||
LastUsed atomic.Int64
|
||
Failures atomic.Uint32
|
||
IsHealthy atomic.Bool
|
||
mu sync.Mutex
|
||
}
|
||
|
||
// NetworkReplicator управляет сетевой репликацией
|
||
type NetworkReplicator struct {
|
||
config *ReplicationRetryConfig
|
||
workerPool *WorkerPool
|
||
dialTimeout time.Duration
|
||
mu sync.RWMutex
|
||
connections map[string]*ReplicatedConnection
|
||
logger LoggerInterface
|
||
stats NetworkReplicationStats
|
||
}
|
||
|
||
// NewNetworkReplicator создаёт новый сетевой репликатор
|
||
func NewNetworkReplicator(config *ReplicationRetryConfig, workerPool *WorkerPool, logger LoggerInterface) *NetworkReplicator {
|
||
if config == nil {
|
||
config = DefaultReplicationRetryConfig()
|
||
}
|
||
// Игнорируем ошибку валидации в production
|
||
validateConfig(config)
|
||
|
||
return &NetworkReplicator{
|
||
config: config,
|
||
workerPool: workerPool,
|
||
dialTimeout: 10 * time.Second,
|
||
connections: make(map[string]*ReplicatedConnection),
|
||
logger: logger,
|
||
}
|
||
}
|
||
|
||
// calculateBackoff вычисляет время задержки с бэкоффом
|
||
func (nr *NetworkReplicator) calculateBackoff(attempt int) time.Duration {
|
||
backoff := float64(nr.config.InitialBackoff) * math.Pow(nr.config.BackoffFactor, float64(attempt))
|
||
if backoff > float64(nr.config.MaxBackoff) {
|
||
backoff = float64(nr.config.MaxBackoff)
|
||
}
|
||
|
||
duration := time.Duration(backoff)
|
||
|
||
if nr.config.JitterEnabled {
|
||
// Добавляем случайный джиттер ±25%
|
||
jitter := time.Duration(float64(duration) * 0.25)
|
||
jitterDelta := int64(jitter) - int64(jitter/2)
|
||
if jitterDelta < 0 {
|
||
jitterDelta = 0
|
||
}
|
||
duration = duration + time.Duration(jitterDelta)
|
||
}
|
||
|
||
return duration
|
||
}
|
||
|
||
// randUint64 возвращает случайное 64-битное число
|
||
func randUint64() uint64 {
|
||
var buf [8]byte
|
||
rand.Read(buf[:])
|
||
return binary.LittleEndian.Uint64(buf[:])
|
||
}
|
||
|
||
// randInt64 возвращает случайное число между min и max
|
||
func randInt64(min, max int64) int64 {
|
||
if min >= max {
|
||
return min
|
||
}
|
||
delta := max - min
|
||
if delta == 0 {
|
||
return min
|
||
}
|
||
n := int64(randUint64() % uint64(delta))
|
||
return min + n
|
||
}
|
||
|
||
// randInt возвращает случайное целое
|
||
func randInt(min, max int) int {
|
||
if min >= max {
|
||
return min
|
||
}
|
||
delta := max - min
|
||
if delta == 0 {
|
||
return min
|
||
}
|
||
n := int(randUint64() % uint64(delta))
|
||
return min + n
|
||
}
|
||
|
||
// Connect устанавливает соединение с узлом
|
||
func (nr *NetworkReplicator) Connect(nodeID, address string) error {
|
||
nr.mu.Lock()
|
||
defer nr.mu.Unlock()
|
||
|
||
if conn, exists := nr.connections[nodeID]; exists {
|
||
if conn.IsHealthy.Load() {
|
||
return nil
|
||
}
|
||
conn.Conn.Close()
|
||
}
|
||
|
||
conn, err := net.DialTimeout("tcp", address, nr.dialTimeout)
|
||
if err != nil {
|
||
return fmt.Errorf("failed to connect to %s: %v", address, err)
|
||
}
|
||
|
||
replicatedConn := &ReplicatedConnection{
|
||
NodeID: nodeID,
|
||
Address: address,
|
||
Conn: conn,
|
||
IsHealthy: atomic.Bool{},
|
||
}
|
||
replicatedConn.IsHealthy.Store(true)
|
||
replicatedConn.LastUsed.Store(time.Now().UnixMilli())
|
||
|
||
nr.connections[nodeID] = replicatedConn
|
||
|
||
if nr.logger != nil {
|
||
nr.logger.Debug(fmt.Sprintf("Connected to node %s at %s", nodeID, address))
|
||
}
|
||
|
||
return nil
|
||
}
|
||
|
||
// sendReplicationRequest отправляет запрос на репликацию
|
||
func (nr *NetworkReplicator) sendReplicationRequest(conn net.Conn, data []byte) error {
|
||
// Формируем сообщение: длина + данные
|
||
lenBuf := make([]byte, 4)
|
||
binary.BigEndian.PutUint32(lenBuf, uint32(len(data)))
|
||
|
||
if _, err := conn.Write(lenBuf); err != nil {
|
||
return fmt.Errorf("failed to write length: %v", err)
|
||
}
|
||
|
||
if _, err := conn.Write(data); err != nil {
|
||
return fmt.Errorf("failed to write data: %v", err)
|
||
}
|
||
|
||
return nil
|
||
}
|
||
|
||
// readReplicationAck читает подтверждение репликации
|
||
func (nr *NetworkReplicator) readReplicationAck(conn net.Conn) error {
|
||
lenBuf := make([]byte, 4)
|
||
if _, err := io.ReadFull(conn, lenBuf); err != nil {
|
||
return fmt.Errorf("failed to read ack length: %v", err)
|
||
}
|
||
|
||
ackLen := binary.BigEndian.Uint32(lenBuf)
|
||
if ackLen > 1024 {
|
||
return fmt.Errorf("ack too large: %d", ackLen)
|
||
}
|
||
|
||
ackData := make([]byte, ackLen)
|
||
if _, err := io.ReadFull(conn, ackData); err != nil {
|
||
return fmt.Errorf("failed to read ack data: %v", err)
|
||
}
|
||
|
||
var ack struct {
|
||
Status string `json:"status"`
|
||
Error string `json:"error,omitempty"`
|
||
}
|
||
if err := json.Unmarshal(ackData, &ack); err != nil {
|
||
return fmt.Errorf("failed to parse ack: %v", err)
|
||
}
|
||
|
||
if ack.Status != "ok" {
|
||
return fmt.Errorf("negative ack: %s", ack.Error)
|
||
}
|
||
|
||
return nil
|
||
}
|
||
|
||
// doSendReplication выполняет отправку репликации
|
||
func (nr *NetworkReplicator) doSendReplication(nodeID, address string, data []byte) error {
|
||
conn := nr.getConnection(nodeID)
|
||
if conn == nil {
|
||
// Пытаемся подключиться
|
||
if err := nr.Connect(nodeID, address); err != nil {
|
||
return err
|
||
}
|
||
conn = nr.getConnection(nodeID)
|
||
if conn == nil {
|
||
return fmt.Errorf("failed to get connection for node %s", nodeID)
|
||
}
|
||
}
|
||
|
||
conn.mu.Lock()
|
||
defer conn.mu.Unlock()
|
||
|
||
if conn.Conn == nil {
|
||
return fmt.Errorf("connection is nil for node %s", nodeID)
|
||
}
|
||
|
||
// Устанавливаем таймаут
|
||
if err := conn.Conn.SetWriteDeadline(time.Now().Add(30 * time.Second)); err != nil {
|
||
return fmt.Errorf("failed to set write deadline: %v", err)
|
||
}
|
||
|
||
// Отправляем данные
|
||
if err := nr.sendReplicationRequest(conn.Conn, data); err != nil {
|
||
conn.IsHealthy.Store(false)
|
||
conn.Conn.Close()
|
||
conn.Conn = nil
|
||
return fmt.Errorf("write failed: %v", err)
|
||
}
|
||
|
||
// Читаем подтверждение
|
||
if err := conn.Conn.SetReadDeadline(time.Now().Add(10 * time.Second)); err != nil {
|
||
return fmt.Errorf("failed to set read deadline: %v", err)
|
||
}
|
||
|
||
if err := nr.readReplicationAck(conn.Conn); err != nil {
|
||
conn.IsHealthy.Store(false)
|
||
return fmt.Errorf("read ack failed: %v", err)
|
||
}
|
||
|
||
conn.IsHealthy.Store(true)
|
||
conn.LastUsed.Store(time.Now().UnixMilli())
|
||
conn.Failures.Store(0)
|
||
|
||
return nil
|
||
}
|
||
|
||
// Replicate отправляет запрос на репликацию с повторными попытками
|
||
func (nr *NetworkReplicator) Replicate(targetNodeID, targetAddress string, data []byte) error {
|
||
startTime := time.Now()
|
||
nr.stats.TotalRequests.Add(1)
|
||
|
||
var lastErr error
|
||
|
||
for attempt := 0; attempt <= nr.config.MaxRetries; attempt++ {
|
||
if attempt > 0 {
|
||
nr.stats.RetriedReqs.Add(1)
|
||
backoff := nr.calculateBackoff(attempt - 1)
|
||
if nr.logger != nil {
|
||
nr.logger.Debug(fmt.Sprintf("Replication retry %d for node %s after %v", attempt, targetNodeID, backoff))
|
||
}
|
||
time.Sleep(backoff)
|
||
}
|
||
|
||
err := nr.doSendReplication(targetNodeID, targetAddress, data)
|
||
if err == nil {
|
||
latency := time.Since(startTime).Milliseconds()
|
||
nr.stats.SuccessfulReqs.Add(1)
|
||
nr.updateAvgLatency(latency)
|
||
nr.stats.BytesSent.Add(uint64(len(data)))
|
||
|
||
if nr.logger != nil {
|
||
nr.logger.Debug(fmt.Sprintf("Replication to %s succeeded after %d attempts, latency: %dms",
|
||
targetNodeID, attempt+1, latency))
|
||
}
|
||
return nil
|
||
}
|
||
|
||
lastErr = err
|
||
if nr.logger != nil {
|
||
nr.logger.Warn(fmt.Sprintf("Replication attempt %d to %s failed: %v", attempt+1, targetNodeID, err))
|
||
}
|
||
|
||
// Помечаем соединение как нездоровое
|
||
nr.markUnhealthy(targetNodeID)
|
||
}
|
||
|
||
nr.stats.FailedReqs.Add(1)
|
||
return fmt.Errorf("replication failed after %d attempts: %v", nr.config.MaxRetries+1, lastErr)
|
||
}
|
||
|
||
// getConnection возвращает соединение с узлом
|
||
func (nr *NetworkReplicator) getConnection(nodeID string) *ReplicatedConnection {
|
||
nr.mu.RLock()
|
||
defer nr.mu.RUnlock()
|
||
return nr.connections[nodeID]
|
||
}
|
||
|
||
// markUnhealthy помечает соединение как нездоровое
|
||
func (nr *NetworkReplicator) markUnhealthy(nodeID string) {
|
||
nr.mu.RLock()
|
||
conn, exists := nr.connections[nodeID]
|
||
nr.mu.RUnlock()
|
||
|
||
if exists {
|
||
conn.IsHealthy.Store(false)
|
||
failures := conn.Failures.Add(1)
|
||
if failures > 3 {
|
||
if nr.logger != nil {
|
||
nr.logger.Warn(fmt.Sprintf("Node %s marked as unhealthy after %d failures", nodeID, failures))
|
||
}
|
||
}
|
||
}
|
||
}
|
||
|
||
// updateAvgLatency обновляет среднюю задержку
|
||
func (nr *NetworkReplicator) updateAvgLatency(latencyMs int64) {
|
||
current := nr.stats.AvgLatencyMs.Load()
|
||
successful := nr.stats.SuccessfulReqs.Load()
|
||
if successful > 0 {
|
||
newAvg := (current*uint64(successful-1) + uint64(latencyMs)) / uint64(successful)
|
||
nr.stats.AvgLatencyMs.Store(newAvg)
|
||
} else {
|
||
nr.stats.AvgLatencyMs.Store(uint64(latencyMs))
|
||
}
|
||
}
|
||
|
||
// ReplicateDocumentAsync асинхронно реплицирует документ
|
||
func (nr *NetworkReplicator) ReplicateDocumentAsync(docID string, targetNodes []*NodeInfo, data []byte) {
|
||
for _, node := range targetNodes {
|
||
targetNodeID := node.ID
|
||
targetAddress := fmt.Sprintf("%s:%d", node.IP, node.Port)
|
||
|
||
err := nr.workerPool.SubmitFunc(fmt.Sprintf("replicate_%s_to_%s", docID, targetNodeID), func() error {
|
||
return nr.Replicate(targetNodeID, targetAddress, data)
|
||
})
|
||
|
||
if err != nil {
|
||
if nr.logger != nil {
|
||
nr.logger.Error(fmt.Sprintf("Failed to submit replication task for %s to %s: %v", docID, targetNodeID, err))
|
||
}
|
||
}
|
||
}
|
||
}
|
||
|
||
// GetStats возвращает статистику репликации
|
||
func (nr *NetworkReplicator) GetStats() map[string]interface{} {
|
||
return map[string]interface{}{
|
||
"total_requests": nr.stats.TotalRequests.Load(),
|
||
"successful": nr.stats.SuccessfulReqs.Load(),
|
||
"failed": nr.stats.FailedReqs.Load(),
|
||
"retried": nr.stats.RetriedReqs.Load(),
|
||
"avg_latency_ms": nr.stats.AvgLatencyMs.Load(),
|
||
"bytes_sent": nr.stats.BytesSent.Load(),
|
||
"bytes_received": nr.stats.BytesReceived.Load(),
|
||
}
|
||
}
|
||
|
||
// Close закрывает все соединения
|
||
func (nr *NetworkReplicator) Close() error {
|
||
nr.mu.Lock()
|
||
defer nr.mu.Unlock()
|
||
|
||
for _, conn := range nr.connections {
|
||
if conn.Conn != nil {
|
||
conn.Conn.Close()
|
||
}
|
||
}
|
||
nr.connections = make(map[string]*ReplicatedConnection)
|
||
return nil
|
||
}
|