feat: add S3 backup integration, keepalive pings, and connection stability

- Add S3 backup manager (pkg/backup/s3.go) for periodic encrypted BadgerDB
  backups with upload to S3-compatible storage (MinIO)
- Add 30s keepalive pings in transport layer to prevent idle read timeouts
- Increase read timeout from 30s to 5 minutes
- Add connection tie-breaking for simultaneous peer connections (lower node
  ID keeps outbound) to prevent connection storms
- Add pointer-specific peer removal (removePeerConn) to prevent orphaned
  goroutines from removing replacement connections
- Export BadgerBackupExecutor type for use by backup package
- Add /backup HTTP endpoint for on-demand backup triggers
- Bump version to 0.3.2
This commit is contained in:
2026-02-27 21:59:49 -08:00
parent 7bc26bfc78
commit db4c7ea771
8 changed files with 414 additions and 30 deletions
+3 -3
View File
@@ -14,9 +14,9 @@ RUN go mod download
# Copy source code
COPY . .
# Build the binaries
RUN go build -o mpcd ./cmd/mpcd
RUN go build -o lux-mpc-cli ./cmd/lux-mpc-cli
# Build the binaries (CGO_ENABLED=0 for static, GOFIPS140=off to avoid QEMU FIPS issue)
RUN CGO_ENABLED=0 GOFIPS140=off go build -o mpcd ./cmd/mpcd
RUN CGO_ENABLED=0 GOFIPS140=off go build -o lux-mpc-cli ./cmd/lux-mpc-cli
# Runtime stage
FROM alpine:latest
+89 -1
View File
@@ -6,6 +6,7 @@ import (
"encoding/hex"
"encoding/json"
"fmt"
"net/http"
"os"
"os/signal"
"path/filepath"
@@ -20,6 +21,7 @@ import (
"github.com/urfave/cli/v3"
"golang.org/x/term"
"github.com/luxfi/mpc/pkg/backup"
"github.com/luxfi/mpc/pkg/config"
"github.com/luxfi/mpc/pkg/constant"
"github.com/luxfi/mpc/pkg/event"
@@ -36,7 +38,7 @@ import (
)
const (
Version = "0.3.1"
Version = "0.3.2"
DefaultBackupPeriodSeconds = 300 // (5 minutes)
)
@@ -698,6 +700,92 @@ func runNodeConsensus(ctx context.Context, c *cli.Command) error {
}
logger.Info("[READY] Node is ready (consensus mode)", "nodeID", nodeID)
// Start HTTP API server
apiAddr := c.String("api")
if apiAddr != "" {
mux := http.NewServeMux()
mux.HandleFunc("/health", func(w http.ResponseWriter, r *http.Request) {
ready := peerRegistry.ArePeersReady()
connected := factory.Transport().GetPeers()
status := "healthy"
httpCode := http.StatusOK
if !ready {
status = "degraded"
httpCode = http.StatusServiceUnavailable
}
w.Header().Set("Content-Type", "application/json")
w.WriteHeader(httpCode)
resp := map[string]interface{}{
"status": status,
"nodeID": nodeID,
"mode": "consensus",
"expectedPeers": len(peerIDs),
"connectedPeers": connected,
"ready": ready,
"threshold": threshold,
"version": Version,
}
json.NewEncoder(w).Encode(resp)
})
mux.HandleFunc("/keys", func(w http.ResponseWriter, r *http.Request) {
keys, err := factory.KeyInfoStore().ListKeys()
w.Header().Set("Content-Type", "application/json")
if err != nil {
w.WriteHeader(http.StatusInternalServerError)
json.NewEncoder(w).Encode(map[string]string{"error": err.Error()})
return
}
json.NewEncoder(w).Encode(keys)
})
mux.HandleFunc("/backup", func(w http.ResponseWriter, r *http.Request) {
if r.Method != http.MethodPost {
w.WriteHeader(http.StatusMethodNotAllowed)
return
}
w.Header().Set("Content-Type", "application/json")
if badgerKV, ok := factory.KVStore().(*kvstore.BadgerKVStore); ok && badgerKV.BackupExecutor != nil {
s3Cfg := backup.S3ConfigFromEnv(nodeID)
mgr, err := backup.NewManager(badgerKV.BackupExecutor, filepath.Join(dataDir, "backups"), nodeID, 0, s3Cfg)
if err != nil {
w.WriteHeader(http.StatusInternalServerError)
json.NewEncoder(w).Encode(map[string]string{"error": err.Error()})
return
}
if err := mgr.RunBackup(); err != nil {
w.WriteHeader(http.StatusInternalServerError)
json.NewEncoder(w).Encode(map[string]string{"error": err.Error()})
return
}
json.NewEncoder(w).Encode(map[string]string{"status": "backup completed"})
} else {
w.WriteHeader(http.StatusServiceUnavailable)
json.NewEncoder(w).Encode(map[string]string{"error": "backup not available"})
}
})
srv := &http.Server{Addr: apiAddr, Handler: mux}
go func() {
logger.Info("HTTP API server starting", "addr", apiAddr)
if err := srv.ListenAndServe(); err != nil && err != http.ErrServerClosed {
logger.Error("HTTP API server failed", err)
}
}()
defer srv.Close()
}
// Start periodic backup with optional S3 upload
backupDir := filepath.Join(dataDir, "backups")
if badgerKV, ok := factory.KVStore().(*kvstore.BadgerKVStore); ok && badgerKV.BackupExecutor != nil {
s3Cfg := backup.S3ConfigFromEnv(nodeID)
backupMgr, err := backup.NewManager(badgerKV.BackupExecutor, backupDir, nodeID, 5*time.Minute, s3Cfg)
if err != nil {
logger.Warn("Failed to create backup manager", "err", err)
} else {
backupMgr.Start()
defer backupMgr.Stop()
logger.Info("Backup manager started", "period", "5m", "s3", s3Cfg != nil)
}
}
// Wait for shutdown signal
sigChan := make(chan os.Signal, 1)
signal.Notify(sigChan, os.Interrupt, syscall.SIGTERM)
+8
View File
@@ -15,6 +15,7 @@ require (
github.com/luxfi/fhe v1.7.6
github.com/luxfi/log v1.4.1
github.com/luxfi/threshold v1.5.5
github.com/minio/minio-go/v7 v7.0.98
github.com/mitchellh/mapstructure v1.5.0
github.com/nats-io/nats.go v1.48.0
github.com/rs/zerolog v1.34.0
@@ -41,6 +42,7 @@ require (
github.com/dustin/go-humanize v1.0.1 // indirect
github.com/fatih/color v1.18.0 // indirect
github.com/fsnotify/fsnotify v1.9.0 // indirect
github.com/go-ini/ini v1.67.0 // indirect
github.com/go-logr/logr v1.4.3 // indirect
github.com/go-logr/stdr v1.2.2 // indirect
github.com/go-viper/mapstructure/v2 v2.5.0 // indirect
@@ -58,26 +60,32 @@ require (
github.com/hashicorp/serf v0.10.2 // indirect
github.com/klauspost/compress v1.18.3 // indirect
github.com/klauspost/cpuid/v2 v2.3.0 // indirect
github.com/klauspost/crc32 v1.3.0 // indirect
github.com/luxfi/lattice/v7 v7.0.0 // indirect
github.com/mattn/go-colorable v0.1.14 // indirect
github.com/mattn/go-isatty v0.0.20 // indirect
github.com/miekg/dns v1.1.66 // indirect
github.com/minio/crc64nvme v1.1.1 // indirect
github.com/minio/md5-simd v1.1.2 // indirect
github.com/mitchellh/go-homedir v1.1.0 // indirect
github.com/munnerz/goautoneg v0.0.0-20191010083416-a7dc8b61c822 // indirect
github.com/nats-io/nkeys v0.4.14 // indirect
github.com/nats-io/nuid v1.0.1 // indirect
github.com/pelletier/go-toml/v2 v2.2.4 // indirect
github.com/philhofer/fwd v1.2.0 // indirect
github.com/pkg/errors v0.9.1 // indirect
github.com/pmezard/go-difflib v1.0.1-0.20181226105442-5d4384ee4fb2 // indirect
github.com/prometheus/client_golang v1.23.2 // indirect
github.com/prometheus/client_model v0.6.2 // indirect
github.com/prometheus/common v0.67.5 // indirect
github.com/prometheus/procfs v0.19.2 // indirect
github.com/rs/xid v1.6.0 // indirect
github.com/sagikazarmark/locafero v0.12.0 // indirect
github.com/spf13/afero v1.15.0 // indirect
github.com/spf13/cast v1.10.0 // indirect
github.com/spf13/pflag v1.0.10 // indirect
github.com/subosito/gotenv v1.6.0 // indirect
github.com/tinylib/msgp v1.6.1 // indirect
github.com/x448/float16 v0.8.4 // indirect
github.com/zeebo/blake3 v0.2.4 // indirect
go.opentelemetry.io/auto/sdk v1.2.1 // indirect
+16
View File
@@ -62,6 +62,8 @@ github.com/fsnotify/fsnotify v1.9.0 h1:2Ml+OJNzbYCTzsxtv8vKSFD9PbJjmhYF14k/jKC7S
github.com/fsnotify/fsnotify v1.9.0/go.mod h1:8jBTzvmWwFyi3Pb8djgCCO5IBqzKJ/Jwo8TRcHyHii0=
github.com/fxamacker/cbor/v2 v2.9.0 h1:NpKPmjDBgUfBms6tr6JZkTHtfFGcMKsw3eGcmD/sapM=
github.com/fxamacker/cbor/v2 v2.9.0/go.mod h1:vM4b+DJCtHn+zz7h3FFp/hDAI9WNWCsZj23V5ytsSxQ=
github.com/go-ini/ini v1.67.0 h1:z6ZrTEZqSWOTyH2FlglNbNgARyHG8oLW9gMELqKr06A=
github.com/go-ini/ini v1.67.0/go.mod h1:ByCAeIL28uOIIG0E3PJtZPDL8WnHpFKFOtgjp+3Ies8=
github.com/go-kit/kit v0.8.0/go.mod h1:xBxKIO96dXMWWy0MnWVtmwkA9/13aqxPnvrjFYMA2as=
github.com/go-kit/kit v0.9.0/go.mod h1:xBxKIO96dXMWWy0MnWVtmwkA9/13aqxPnvrjFYMA2as=
github.com/go-kit/log v0.1.0/go.mod h1:zbhenjAZHb184qTLMA9ZjW7ThYL0H2mk7Q6pNt4vbaY=
@@ -154,8 +156,11 @@ github.com/julienschmidt/httprouter v1.2.0/go.mod h1:SYymIcj16QtmaHHD7aYtjjsJG7V
github.com/julienschmidt/httprouter v1.3.0/go.mod h1:JR6WtHb+2LUe8TCKY3cZOxFyyO8IZAc4RVcycCCAKdM=
github.com/klauspost/compress v1.18.3 h1:9PJRvfbmTabkOX8moIpXPbMMbYN60bWImDDU7L+/6zw=
github.com/klauspost/compress v1.18.3/go.mod h1:R0h/fSBs8DE4ENlcrlib3PsXS61voFxhIs2DeRhCvJ4=
github.com/klauspost/cpuid/v2 v2.0.1/go.mod h1:FInQzS24/EEf25PyTYn52gqo7WaD8xa0213Md/qVLRg=
github.com/klauspost/cpuid/v2 v2.3.0 h1:S4CRMLnYUhGeDFDqkGriYKdfoFlDnMtqTiI/sFzhA9Y=
github.com/klauspost/cpuid/v2 v2.3.0/go.mod h1:hqwkgyIinND0mEev00jJYCxPNVRVXFQeu1XKlok6oO0=
github.com/klauspost/crc32 v1.3.0 h1:sSmTt3gUt81RP655XGZPElI0PelVTZ6YwCRnPSupoFM=
github.com/klauspost/crc32 v1.3.0/go.mod h1:D7kQaZhnkX/Y0tstFGf8VUzv2UofNGqCjnC3zdHB0Hw=
github.com/konsorten/go-windows-terminal-sequences v1.0.1/go.mod h1:T0+1ngSBFLxvqU3pZ+m/2kptfBszLMUkC4ZK/EgS/cQ=
github.com/konsorten/go-windows-terminal-sequences v1.0.3/go.mod h1:T0+1ngSBFLxvqU3pZ+m/2kptfBszLMUkC4ZK/EgS/cQ=
github.com/kr/logfmt v0.0.0-20140226030751-b84e30acd515/go.mod h1:+0opPa2QZZtGFBFZlji/RkVcI2GknAs/DXo4wKdlNEc=
@@ -192,6 +197,12 @@ github.com/mattn/go-isatty v0.0.20/go.mod h1:W+V8PltTTMOvKvAeJH7IuucS94S2C6jfK/D
github.com/matttproud/golang_protobuf_extensions v1.0.1/go.mod h1:D8He9yQNgCq6Z5Ld7szi9bcBfOoFv/3dc6xSMkL2PC0=
github.com/miekg/dns v1.1.66 h1:FeZXOS3VCVsKnEAd+wBkjMC3D2K+ww66Cq3VnCINuJE=
github.com/miekg/dns v1.1.66/go.mod h1:jGFzBsSNbJw6z1HYut1RKBKHA9PBdxeHrZG8J+gC2WE=
github.com/minio/crc64nvme v1.1.1 h1:8dwx/Pz49suywbO+auHCBpCtlW1OfpcLN7wYgVR6wAI=
github.com/minio/crc64nvme v1.1.1/go.mod h1:eVfm2fAzLlxMdUGc0EEBGSMmPwmXD5XiNRpnu9J3bvg=
github.com/minio/md5-simd v1.1.2 h1:Gdi1DZK69+ZVMoNHRXJyNcxrMA4dSxoYHZSQbirFg34=
github.com/minio/md5-simd v1.1.2/go.mod h1:MzdKDxYpY2BT9XQFocsiZf/NKVtR7nkE4RoEpN+20RM=
github.com/minio/minio-go/v7 v7.0.98 h1:MeAVKjLVz+XJ28zFcuYyImNSAh8Mq725uNW4beRisi0=
github.com/minio/minio-go/v7 v7.0.98/go.mod h1:cY0Y+W7yozf0mdIclrttzo1Iiu7mEf9y7nk2uXqMOvM=
github.com/mitchellh/go-homedir v1.1.0 h1:lukF9ziXFxDFPkA1vsr5zpc1XuPDn/wFntq5mG+4E0Y=
github.com/mitchellh/go-homedir v1.1.0/go.mod h1:SfyaCUpYCn1Vlf4IUYiD9fPX4A5wJrkLzIz1N1q0pr0=
github.com/mitchellh/mapstructure v1.5.0 h1:jeMsZIYE/09sWLaz43PL7Gy6RuMjD2eJVyuac5Z2hdY=
@@ -218,6 +229,8 @@ github.com/pascaldekloe/goe v0.1.0 h1:cBOtyMzM9HTpWjXfbbunk26uA6nG3a8n06Wieeh0Mw
github.com/pascaldekloe/goe v0.1.0/go.mod h1:lzWF7FIEvWOWxwDKqyGYQf6ZUaNfKdP144TG7ZOy1lc=
github.com/pelletier/go-toml/v2 v2.2.4 h1:mye9XuhQ6gvn5h28+VilKrrPoQVanw5PMw/TB0t5Ec4=
github.com/pelletier/go-toml/v2 v2.2.4/go.mod h1:2gIqNv+qfxSVS7cM2xJQKtLSTLUE9V8t9Stt+h56mCY=
github.com/philhofer/fwd v1.2.0 h1:e6DnBTl7vGY+Gz322/ASL4Gyp1FspeMvx1RNDoToZuM=
github.com/philhofer/fwd v1.2.0/go.mod h1:RqIHx9QI14HlwKwm98g9Re5prTQ6LdeRQn+gXJFxsJM=
github.com/pkg/errors v0.8.0/go.mod h1:bwawxfHBFNV+L2hUp1rHADufV3IMtnDRdf1r5NINEl0=
github.com/pkg/errors v0.8.1/go.mod h1:bwawxfHBFNV+L2hUp1rHADufV3IMtnDRdf1r5NINEl0=
github.com/pkg/errors v0.9.1 h1:FEBLx1zS214owpjy7qsBeixbURkuhQAwrK5UwLGTwt4=
@@ -252,6 +265,7 @@ github.com/prometheus/procfs v0.19.2 h1:zUMhqEW66Ex7OXIiDkll3tl9a1ZdilUOd/F6ZXw4
github.com/prometheus/procfs v0.19.2/go.mod h1:M0aotyiemPhBCM0z5w87kL22CxfcH05ZpYlu+b4J7mw=
github.com/rogpeppe/go-internal v1.14.1 h1:UQB4HGPB6osV0SQTLymcB4TgvyWu6ZyliaW0tI/otEQ=
github.com/rogpeppe/go-internal v1.14.1/go.mod h1:MaRKkUm5W0goXpeCfT7UZI6fk/L7L7so1lCWt35ZSgc=
github.com/rs/xid v1.6.0 h1:fV591PaemRlL6JfRxGDEPl69wICngIQ3shQtzfy2gxU=
github.com/rs/xid v1.6.0/go.mod h1:7XoLgs4eV+QndskICGsho+ADou8ySMSjJKDIan90Nz0=
github.com/rs/zerolog v1.34.0 h1:k43nTLIwcTVQAncfCw4KZ2VY6ukYoZaBPNOE8txlOeY=
github.com/rs/zerolog v1.34.0/go.mod h1:bJsvje4Z08ROH4Nhs5iH600c3IkWhwp44iRc54W6wYQ=
@@ -284,6 +298,8 @@ github.com/stretchr/testify v1.11.1 h1:7s2iGBzp5EwR7/aIZr8ao5+dra3wiQyKjjFuvgVKu
github.com/stretchr/testify v1.11.1/go.mod h1:wZwfW3scLgRK+23gO65QZefKpKQRnfz6sD981Nm4B6U=
github.com/subosito/gotenv v1.6.0 h1:9NlTDc1FTs4qu0DDq7AEtTPNw6SVm7uBMsUCUjABIf8=
github.com/subosito/gotenv v1.6.0/go.mod h1:Dk4QP5c2W3ibzajGcXpNraDfq2IrhjMIvMSWPKKo0FU=
github.com/tinylib/msgp v1.6.1 h1:ESRv8eL3u+DNHUoSAAQRE50Hm162zqAnBoGv9PzScPY=
github.com/tinylib/msgp v1.6.1/go.mod h1:RSp0LW9oSxFut3KzESt5Voq4GVWyS+PSulT77roAqEA=
github.com/tv42/httpunix v0.0.0-20150427012821-b75d8614f926/go.mod h1:9ESjWnEqriFuLhtthL60Sar/7RFoluCcXsuvEwTV5KM=
github.com/urfave/cli/v3 v3.6.2 h1:lQuqiPrZ1cIz8hz+HcrG0TNZFxU70dPZ3Yl+pSrH9A8=
github.com/urfave/cli/v3 v3.6.2/go.mod h1:ysVLtOEmg2tOy6PknnYVhDoouyC/6N42TMeoMzskhso=
+211
View File
@@ -0,0 +1,211 @@
package backup
import (
"context"
"fmt"
"os"
"path/filepath"
"strings"
"time"
"github.com/minio/minio-go/v7"
"github.com/minio/minio-go/v7/pkg/credentials"
"github.com/luxfi/mpc/pkg/kvstore"
"github.com/luxfi/mpc/pkg/logger"
)
// S3Config configures S3-compatible backup storage
type S3Config struct {
Endpoint string // e.g., "s3.amazonaws.com" or "minio.example.com"
Bucket string // e.g., "lux-mpc-backups"
AccessKey string
SecretKey string
Region string // e.g., "us-east-1"
UseSSL bool
Prefix string // e.g., "mpc/node0/"
}
// S3ConfigFromEnv creates S3Config from environment variables
func S3ConfigFromEnv(nodeID string) *S3Config {
endpoint := os.Getenv("S3_ENDPOINT")
if endpoint == "" {
return nil // S3 not configured
}
bucket := os.Getenv("S3_BUCKET")
if bucket == "" {
bucket = "lux-mpc-backups"
}
prefix := os.Getenv("S3_PREFIX")
if prefix == "" {
prefix = fmt.Sprintf("mpc/%s/", nodeID)
}
if !strings.HasSuffix(prefix, "/") {
prefix += "/"
}
useSSL := os.Getenv("S3_USE_SSL") != "false"
return &S3Config{
Endpoint: endpoint,
Bucket: bucket,
AccessKey: os.Getenv("S3_ACCESS_KEY"),
SecretKey: os.Getenv("S3_SECRET_KEY"),
Region: os.Getenv("S3_REGION"),
UseSSL: useSSL,
Prefix: prefix,
}
}
// Manager handles periodic backups with optional S3 upload
type Manager struct {
executor *kvstore.BadgerBackupExecutor
backupDir string
s3Config *S3Config
s3Client *minio.Client
nodeID string
period time.Duration
done chan struct{}
}
// NewManager creates a backup manager
func NewManager(executor *kvstore.BadgerBackupExecutor, backupDir, nodeID string, period time.Duration, s3Cfg *S3Config) (*Manager, error) {
m := &Manager{
executor: executor,
backupDir: backupDir,
s3Config: s3Cfg,
nodeID: nodeID,
period: period,
done: make(chan struct{}),
}
if s3Cfg != nil && s3Cfg.Endpoint != "" {
client, err := minio.New(s3Cfg.Endpoint, &minio.Options{
Creds: credentials.NewStaticV4(s3Cfg.AccessKey, s3Cfg.SecretKey, ""),
Secure: s3Cfg.UseSSL,
Region: s3Cfg.Region,
})
if err != nil {
return nil, fmt.Errorf("failed to create S3 client: %w", err)
}
m.s3Client = client
// Ensure bucket exists
ctx, cancel := context.WithTimeout(context.Background(), 10*time.Second)
defer cancel()
exists, err := client.BucketExists(ctx, s3Cfg.Bucket)
if err != nil {
logger.Warn("Failed to check S3 bucket", "bucket", s3Cfg.Bucket, "err", err)
} else if !exists {
if err := client.MakeBucket(ctx, s3Cfg.Bucket, minio.MakeBucketOptions{Region: s3Cfg.Region}); err != nil {
logger.Warn("Failed to create S3 bucket", "bucket", s3Cfg.Bucket, "err", err)
} else {
logger.Info("Created S3 bucket", "bucket", s3Cfg.Bucket)
}
}
logger.Info("S3 backup enabled",
"endpoint", s3Cfg.Endpoint,
"bucket", s3Cfg.Bucket,
"prefix", s3Cfg.Prefix,
)
}
return m, nil
}
// Start begins the periodic backup loop
func (m *Manager) Start() {
go m.loop()
}
// Stop stops the backup manager
func (m *Manager) Stop() {
close(m.done)
}
func (m *Manager) loop() {
ticker := time.NewTicker(m.period)
defer ticker.Stop()
for {
select {
case <-m.done:
return
case <-ticker.C:
if err := m.RunBackup(); err != nil {
logger.Error("Backup failed", err)
}
}
}
}
// RunBackup executes a backup and uploads to S3 if configured
func (m *Manager) RunBackup() error {
// Get list of files before backup
beforeFiles := m.executor.SortedEncryptedBackups()
// Execute local backup
if err := m.executor.Execute(); err != nil {
return fmt.Errorf("local backup failed: %w", err)
}
// Find new files
afterFiles := m.executor.SortedEncryptedBackups()
newFiles := findNewFiles(beforeFiles, afterFiles)
if len(newFiles) == 0 {
return nil // No new backup (no changes)
}
// Upload new files to S3
if m.s3Client != nil {
for _, f := range newFiles {
if err := m.uploadToS3(f); err != nil {
logger.Error("S3 upload failed", err, "file", f)
// Don't return error - local backup succeeded
}
}
}
return nil
}
func (m *Manager) uploadToS3(localPath string) error {
ctx, cancel := context.WithTimeout(context.Background(), 5*time.Minute)
defer cancel()
filename := filepath.Base(localPath)
objectName := m.s3Config.Prefix + filename
info, err := m.s3Client.FPutObject(ctx, m.s3Config.Bucket, objectName, localPath, minio.PutObjectOptions{
ContentType: "application/octet-stream",
})
if err != nil {
return fmt.Errorf("S3 upload failed: %w", err)
}
logger.Info("Backup uploaded to S3",
"file", filename,
"bucket", m.s3Config.Bucket,
"object", objectName,
"size", info.Size,
)
return nil
}
func findNewFiles(before, after []string) []string {
existing := make(map[string]bool, len(before))
for _, f := range before {
existing[f] = true
}
var newFiles []string
for _, f := range after {
if !existing[f] {
newFiles = append(newFiles, f)
}
}
return newFiles
}
+1 -1
View File
@@ -17,7 +17,7 @@ var (
// BadgerKVStore is an implementation of the KVStore interface using BadgerDB.
type BadgerKVStore struct {
DB *badger.DB
BackupExecutor *badgerBackupExecutor
BackupExecutor *BadgerBackupExecutor
}
type BadgerConfig struct {
+10 -9
View File
@@ -42,7 +42,8 @@ type BadgerBackupVersionInfo struct {
UpdatedAt string `json:"updated_at"` // RFC3339
}
type badgerBackupExecutor struct {
// BadgerBackupExecutor handles encrypted BadgerDB backups
type BadgerBackupExecutor struct {
NodeID string
DB *badger.DB
BackupEncryptionKey []byte
@@ -55,14 +56,14 @@ func NewBadgerBackupExecutor(
db *badger.DB,
backupEncryptionKey []byte,
backupDir string,
) *badgerBackupExecutor {
) *BadgerBackupExecutor {
if backupDir == "" {
backupDir = defaultBackupDir
}
if err := os.MkdirAll(backupDir, 0700); err != nil {
panic(fmt.Errorf("failed to create backup directory: %w", err))
}
return &badgerBackupExecutor{
return &BadgerBackupExecutor{
NodeID: nodeID,
DB: db,
BackupEncryptionKey: backupEncryptionKey,
@@ -70,7 +71,7 @@ func NewBadgerBackupExecutor(
}
}
func (b *badgerBackupExecutor) Execute() error {
func (b *BadgerBackupExecutor) Execute() error {
info, err := b.LoadVersionInfo()
if err != nil {
return fmt.Errorf("failed to load version info: %w", err)
@@ -142,7 +143,7 @@ func (b *badgerBackupExecutor) Execute() error {
return nil
}
func (b *badgerBackupExecutor) SaveVersionInfo(counter, since uint64) error {
func (b *BadgerBackupExecutor) SaveVersionInfo(counter, since uint64) error {
info := BadgerBackupVersionInfo{
Version: counter,
Since: since,
@@ -156,7 +157,7 @@ func (b *badgerBackupExecutor) SaveVersionInfo(counter, since uint64) error {
return os.WriteFile(versionFile, data, 0600)
}
func (b *badgerBackupExecutor) LoadVersionInfo() (BadgerBackupVersionInfo, error) {
func (b *BadgerBackupExecutor) LoadVersionInfo() (BadgerBackupVersionInfo, error) {
var info BadgerBackupVersionInfo
versionFile := filepath.Join(b.BackupDir, "latest.version")
data, err := os.ReadFile(versionFile)
@@ -175,13 +176,13 @@ func (b *badgerBackupExecutor) LoadVersionInfo() (BadgerBackupVersionInfo, error
return info, err
}
func (b *badgerBackupExecutor) SortedEncryptedBackups() []string {
func (b *BadgerBackupExecutor) SortedEncryptedBackups() []string {
files, _ := filepath.Glob(filepath.Join(b.BackupDir, "backup-*.enc"))
sort.Strings(files)
return files
}
func (b *badgerBackupExecutor) RestoreAllBackupsEncrypted(restorePath string, encryptionKey []byte) error {
func (b *BadgerBackupExecutor) RestoreAllBackupsEncrypted(restorePath string, encryptionKey []byte) error {
err := os.MkdirAll(restorePath, 0700)
if err != nil {
return fmt.Errorf("failed to create restore directory: %w", err)
@@ -214,7 +215,7 @@ func (b *badgerBackupExecutor) RestoreAllBackupsEncrypted(restorePath string, en
return nil
}
func (b *badgerBackupExecutor) loadEncryptedBackup(db *badger.DB, path string) error {
func (b *BadgerBackupExecutor) loadEncryptedBackup(db *badger.DB, path string) error {
f, err := os.Open(path)
if err != nil {
return err
+76 -16
View File
@@ -56,7 +56,7 @@ type Config struct {
// DefaultConfig returns sensible defaults
func DefaultConfig() *Config {
return &Config{
ReadTimeout: 30 * time.Second,
ReadTimeout: 5 * time.Minute,
WriteTimeout: 10 * time.Second,
BufferSize: 64 * 1024,
}
@@ -210,6 +210,15 @@ func (t *Transport) connectToPeer(ctx context.Context, nodeID, addr string) {
default:
}
// Check if we already have a connection (e.g., inbound)
t.peersMu.RLock()
_, exists := t.peers[nodeID]
t.peersMu.RUnlock()
if exists {
time.Sleep(5 * time.Second)
continue
}
conn, err := net.DialTimeout("tcp", addr, 10*time.Second)
if err != nil {
logger.Warn("Failed to connect to peer", "nodeID", nodeID, "addr", addr, "err", err)
@@ -220,7 +229,13 @@ func (t *Transport) connectToPeer(ctx context.Context, nodeID, addr string) {
backoff = time.Second // Reset backoff on success
peer := t.addPeer(nodeID, addr, conn)
peer, accepted := t.addPeerTieBreak(nodeID, addr, conn, true)
if !accepted {
conn.Close()
logger.Debug("Outbound connection rejected (tie-break)", "nodeID", nodeID)
time.Sleep(5 * time.Second)
continue
}
logger.Info("Connected to peer", "nodeID", nodeID, "addr", addr)
// Send our identity
@@ -229,8 +244,8 @@ func (t *Transport) connectToPeer(ctx context.Context, nodeID, addr string) {
// Handle messages from this peer
t.handlePeer(ctx, peer)
// Connection closed, remove peer
t.removePeer(nodeID)
// Connection closed, remove only THIS connection
t.removePeerConn(nodeID, peer)
logger.Warn("Disconnected from peer", "nodeID", nodeID)
}
}
@@ -238,37 +253,68 @@ func (t *Transport) connectToPeer(ctx context.Context, nodeID, addr string) {
// handleIncoming handles an incoming connection
func (t *Transport) handleIncoming(ctx context.Context, conn net.Conn) {
defer t.wg.Done()
defer conn.Close()
// Read identity message first
msgType, payload, err := ReadMessage(conn)
if err != nil {
logger.Error("Failed to read identity", err)
conn.Close()
return
}
if msgType != MsgMPCReady {
logger.Warn("Expected identity message, got", "type", msgType)
conn.Close()
return
}
var ready ReadySignal
if err := json.Unmarshal(payload, &ready); err != nil {
logger.Error("Failed to unmarshal identity", err)
conn.Close()
return
}
peer := t.addPeer(ready.NodeID, conn.RemoteAddr().String(), conn)
peer, accepted := t.addPeerTieBreak(ready.NodeID, conn.RemoteAddr().String(), conn, false)
if !accepted {
logger.Debug("Inbound connection rejected (tie-break)", "nodeID", ready.NodeID)
conn.Close()
return
}
logger.Info("Incoming connection from peer", "nodeID", ready.NodeID)
// Handle messages from this peer
t.handlePeer(ctx, peer)
t.removePeer(ready.NodeID)
t.removePeerConn(ready.NodeID, peer)
}
// handlePeer reads messages from a peer connection
func (t *Transport) handlePeer(ctx context.Context, peer *peerConn) {
// Send keepalive pings every 30s to prevent idle timeout
pingDone := make(chan struct{})
go func() {
ticker := time.NewTicker(30 * time.Second)
defer ticker.Stop()
for {
select {
case <-pingDone:
return
case <-ctx.Done():
return
case <-t.done:
return
case <-ticker.C:
msg := Message{Type: MsgMPCPing, From: t.config.NodeID}
payload, _ := json.Marshal(msg)
if err := peer.send(MsgMPCPing, payload); err != nil {
return
}
}
}
}()
defer close(pingDone)
for {
select {
case <-ctx.Done():
@@ -536,8 +582,10 @@ func (t *Transport) sendPong(to string) {
}
}
// addPeer adds a peer connection
func (t *Transport) addPeer(nodeID, addr string, conn net.Conn) *peerConn {
// addPeerTieBreak adds a peer connection with tie-breaking for simultaneous connections.
// When both sides connect simultaneously, the node with the lower ID keeps its outbound connection.
// Returns the peer and true if accepted, or nil and false if rejected.
func (t *Transport) addPeerTieBreak(nodeID, addr string, conn net.Conn, isOutbound bool) (*peerConn, bool) {
peer := &peerConn{
nodeID: nodeID,
addr: addr,
@@ -547,20 +595,32 @@ func (t *Transport) addPeer(nodeID, addr string, conn net.Conn) *peerConn {
}
t.peersMu.Lock()
// Close existing connection if any
defer t.peersMu.Unlock()
if existing, ok := t.peers[nodeID]; ok {
// Tie-break: lower ID wins outbound
var keepNew bool
if isOutbound {
keepNew = t.config.NodeID < nodeID
} else {
keepNew = nodeID < t.config.NodeID
}
if !keepNew {
return nil, false
}
existing.close()
}
t.peers[nodeID] = peer
t.peersMu.Unlock()
return peer
t.peers[nodeID] = peer
return peer, true
}
// removePeer removes a peer connection
func (t *Transport) removePeer(nodeID string) {
// removePeerConn removes a peer connection only if it matches the given pointer.
// This prevents a replaced connection's cleanup from removing the new connection.
func (t *Transport) removePeerConn(nodeID string, peer *peerConn) {
t.peersMu.Lock()
if peer, ok := t.peers[nodeID]; ok {
if current, ok := t.peers[nodeID]; ok && current == peer {
peer.close()
delete(t.peers, nodeID)
}