feat(federation): add dolt sql-server mode for daemon (bd-wkumz.2)
Add --federation flag to bd daemon start that runs dolt sql-server instead of the embedded driver. Enables multi-writer support and exposes remotesapi on port 8080 for peer-to-peer push/pull. Changes: - Add --federation flag to daemon start command - Create dolt server manager (internal/storage/dolt/server.go) - Update DoltStore to support server mode via MySQL protocol - Integrate server lifecycle into daemon (auto-start/stop) - Add tests for server management and server mode connections Co-Authored-By: Claude Opus 4.5 <noreply@anthropic.com>
This commit is contained in:
beads/crew/darcy
committed by
Steve Yegge
Steve Yegge
parent
458fb7197a
commit
da4584ae57
320
internal/storage/dolt/server.go
Normal file
320
internal/storage/dolt/server.go
Normal file
@@ -0,0 +1,320 @@
|
||||
// Package dolt implements the storage interface using Dolt (versioned MySQL-compatible database).
|
||||
//
|
||||
// This file implements the dolt sql-server management for federation mode.
|
||||
// When federation is enabled, we run dolt sql-server instead of the embedded driver
|
||||
// to enable multi-writer support and expose the remotesapi for peer-to-peer sync.
|
||||
package dolt
|
||||
|
||||
import (
|
||||
"context"
|
||||
"fmt"
|
||||
"net"
|
||||
"os"
|
||||
"os/exec"
|
||||
"path/filepath"
|
||||
"strconv"
|
||||
"strings"
|
||||
"sync"
|
||||
"syscall"
|
||||
"time"
|
||||
)
|
||||
|
||||
const (
|
||||
// DefaultSQLPort is the default port for dolt sql-server MySQL protocol
|
||||
DefaultSQLPort = 3306
|
||||
// DefaultRemotesAPIPort is the default port for dolt remotesapi (peer-to-peer sync)
|
||||
DefaultRemotesAPIPort = 8080
|
||||
// ServerStartTimeout is how long to wait for server to start
|
||||
ServerStartTimeout = 30 * time.Second
|
||||
// ServerStopTimeout is how long to wait for graceful shutdown
|
||||
ServerStopTimeout = 10 * time.Second
|
||||
)
|
||||
|
||||
// ServerConfig holds configuration for the dolt sql-server
|
||||
type ServerConfig struct {
|
||||
DataDir string // Path to Dolt database directory
|
||||
SQLPort int // MySQL protocol port (default: 3306)
|
||||
RemotesAPIPort int // remotesapi port for peer sync (default: 8080)
|
||||
Host string // Host to bind to (default: 127.0.0.1)
|
||||
LogFile string // Log file for server output (optional)
|
||||
User string // MySQL user (default: root)
|
||||
ReadOnly bool // Start in read-only mode
|
||||
}
|
||||
|
||||
// Server manages a dolt sql-server process
|
||||
type Server struct {
|
||||
cfg ServerConfig
|
||||
cmd *exec.Cmd
|
||||
mu sync.Mutex
|
||||
running bool
|
||||
pidFile string
|
||||
}
|
||||
|
||||
// NewServer creates a new dolt sql-server manager
|
||||
func NewServer(cfg ServerConfig) *Server {
|
||||
if cfg.SQLPort == 0 {
|
||||
cfg.SQLPort = DefaultSQLPort
|
||||
}
|
||||
if cfg.RemotesAPIPort == 0 {
|
||||
cfg.RemotesAPIPort = DefaultRemotesAPIPort
|
||||
}
|
||||
if cfg.Host == "" {
|
||||
cfg.Host = "127.0.0.1"
|
||||
}
|
||||
if cfg.User == "" {
|
||||
cfg.User = "root"
|
||||
}
|
||||
return &Server{
|
||||
cfg: cfg,
|
||||
pidFile: filepath.Join(cfg.DataDir, "dolt-server.pid"),
|
||||
}
|
||||
}
|
||||
|
||||
// Start starts the dolt sql-server process
|
||||
func (s *Server) Start(ctx context.Context) error {
|
||||
s.mu.Lock()
|
||||
defer s.mu.Unlock()
|
||||
|
||||
if s.running {
|
||||
return fmt.Errorf("server already running")
|
||||
}
|
||||
|
||||
// Check if ports are available
|
||||
if err := s.checkPortAvailable(s.cfg.SQLPort); err != nil {
|
||||
return fmt.Errorf("SQL port %d not available: %w", s.cfg.SQLPort, err)
|
||||
}
|
||||
if err := s.checkPortAvailable(s.cfg.RemotesAPIPort); err != nil {
|
||||
return fmt.Errorf("remotesapi port %d not available: %w", s.cfg.RemotesAPIPort, err)
|
||||
}
|
||||
|
||||
// Build command args
|
||||
// Note: --user was removed in recent dolt versions, users are created with CREATE USER
|
||||
args := []string{
|
||||
"sql-server",
|
||||
"--host", s.cfg.Host,
|
||||
"--port", strconv.Itoa(s.cfg.SQLPort),
|
||||
"--remotesapi-port", strconv.Itoa(s.cfg.RemotesAPIPort),
|
||||
"--no-auto-commit", // Let the application manage commits
|
||||
}
|
||||
|
||||
if s.cfg.ReadOnly {
|
||||
args = append(args, "--readonly")
|
||||
}
|
||||
|
||||
// Create command
|
||||
s.cmd = exec.CommandContext(ctx, "dolt", args...)
|
||||
s.cmd.Dir = s.cfg.DataDir
|
||||
|
||||
// Set up process group for clean shutdown
|
||||
s.cmd.SysProcAttr = &syscall.SysProcAttr{
|
||||
Setpgid: true,
|
||||
}
|
||||
|
||||
// Set up logging
|
||||
if s.cfg.LogFile != "" {
|
||||
logFile, err := os.OpenFile(s.cfg.LogFile, os.O_CREATE|os.O_WRONLY|os.O_APPEND, 0600)
|
||||
if err != nil {
|
||||
return fmt.Errorf("failed to open log file: %w", err)
|
||||
}
|
||||
s.cmd.Stdout = logFile
|
||||
s.cmd.Stderr = logFile
|
||||
} else {
|
||||
// Discard output if no log file specified
|
||||
s.cmd.Stdout = nil
|
||||
s.cmd.Stderr = nil
|
||||
}
|
||||
|
||||
// Start the server
|
||||
if err := s.cmd.Start(); err != nil {
|
||||
return fmt.Errorf("failed to start dolt sql-server: %w", err)
|
||||
}
|
||||
|
||||
// Write PID file
|
||||
if err := os.WriteFile(s.pidFile, []byte(strconv.Itoa(s.cmd.Process.Pid)), 0600); err != nil {
|
||||
// Non-fatal, just log
|
||||
fmt.Fprintf(os.Stderr, "Warning: failed to write dolt server PID file: %v\n", err)
|
||||
}
|
||||
|
||||
// Wait for server to be ready
|
||||
if err := s.waitForReady(ctx); err != nil {
|
||||
// Server failed to start, clean up
|
||||
_ = s.cmd.Process.Kill()
|
||||
_ = os.Remove(s.pidFile)
|
||||
return fmt.Errorf("server failed to become ready: %w", err)
|
||||
}
|
||||
|
||||
s.running = true
|
||||
return nil
|
||||
}
|
||||
|
||||
// Stop stops the dolt sql-server process gracefully
|
||||
func (s *Server) Stop() error {
|
||||
s.mu.Lock()
|
||||
defer s.mu.Unlock()
|
||||
|
||||
if !s.running || s.cmd == nil || s.cmd.Process == nil {
|
||||
return nil
|
||||
}
|
||||
|
||||
// Try graceful shutdown first (SIGTERM)
|
||||
if err := s.cmd.Process.Signal(syscall.SIGTERM); err != nil {
|
||||
// Process may already be dead
|
||||
if !strings.Contains(err.Error(), "process already finished") {
|
||||
return fmt.Errorf("failed to send SIGTERM: %w", err)
|
||||
}
|
||||
}
|
||||
|
||||
// Wait for graceful shutdown with timeout
|
||||
done := make(chan error, 1)
|
||||
go func() {
|
||||
_, err := s.cmd.Process.Wait()
|
||||
done <- err
|
||||
}()
|
||||
|
||||
select {
|
||||
case <-done:
|
||||
// Process exited
|
||||
case <-time.After(ServerStopTimeout):
|
||||
// Force kill
|
||||
_ = s.cmd.Process.Kill()
|
||||
<-done // Wait for process to be reaped
|
||||
}
|
||||
|
||||
// Clean up PID file
|
||||
_ = os.Remove(s.pidFile)
|
||||
s.running = false
|
||||
s.cmd = nil
|
||||
|
||||
return nil
|
||||
}
|
||||
|
||||
// IsRunning returns true if the server is running
|
||||
func (s *Server) IsRunning() bool {
|
||||
s.mu.Lock()
|
||||
defer s.mu.Unlock()
|
||||
return s.running
|
||||
}
|
||||
|
||||
// SQLPort returns the SQL port
|
||||
func (s *Server) SQLPort() int {
|
||||
return s.cfg.SQLPort
|
||||
}
|
||||
|
||||
// RemotesAPIPort returns the remotesapi port
|
||||
func (s *Server) RemotesAPIPort() int {
|
||||
return s.cfg.RemotesAPIPort
|
||||
}
|
||||
|
||||
// Host returns the host
|
||||
func (s *Server) Host() string {
|
||||
return s.cfg.Host
|
||||
}
|
||||
|
||||
// DSN returns the MySQL DSN for connecting to the server
|
||||
func (s *Server) DSN(database string) string {
|
||||
return fmt.Sprintf("%s@tcp(%s:%d)/%s",
|
||||
s.cfg.User, s.cfg.Host, s.cfg.SQLPort, database)
|
||||
}
|
||||
|
||||
// checkPortAvailable checks if a TCP port is available
|
||||
func (s *Server) checkPortAvailable(port int) error {
|
||||
addr := fmt.Sprintf("%s:%d", s.cfg.Host, port)
|
||||
listener, err := net.Listen("tcp", addr)
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
_ = listener.Close()
|
||||
return nil
|
||||
}
|
||||
|
||||
// waitForReady waits for the server to accept connections
|
||||
func (s *Server) waitForReady(ctx context.Context) error {
|
||||
deadline := time.Now().Add(ServerStartTimeout)
|
||||
addr := fmt.Sprintf("%s:%d", s.cfg.Host, s.cfg.SQLPort)
|
||||
|
||||
for time.Now().Before(deadline) {
|
||||
select {
|
||||
case <-ctx.Done():
|
||||
return ctx.Err()
|
||||
default:
|
||||
}
|
||||
|
||||
// Check if process is still alive using signal 0
|
||||
if s.cmd.Process != nil {
|
||||
if err := s.cmd.Process.Signal(syscall.Signal(0)); err != nil {
|
||||
return fmt.Errorf("server process exited unexpectedly")
|
||||
}
|
||||
}
|
||||
|
||||
// Try to connect
|
||||
conn, err := net.DialTimeout("tcp", addr, 1*time.Second)
|
||||
if err == nil {
|
||||
_ = conn.Close()
|
||||
return nil
|
||||
}
|
||||
|
||||
time.Sleep(100 * time.Millisecond)
|
||||
}
|
||||
|
||||
return fmt.Errorf("timeout waiting for server to start on %s", addr)
|
||||
}
|
||||
|
||||
// GetRunningServerPID returns the PID of a running server from the PID file, or 0 if not running
|
||||
func GetRunningServerPID(dataDir string) int {
|
||||
pidFile := filepath.Join(dataDir, "dolt-server.pid")
|
||||
data, err := os.ReadFile(pidFile)
|
||||
if err != nil {
|
||||
return 0
|
||||
}
|
||||
|
||||
pid, err := strconv.Atoi(strings.TrimSpace(string(data)))
|
||||
if err != nil {
|
||||
return 0
|
||||
}
|
||||
|
||||
// Check if process is actually running
|
||||
process, err := os.FindProcess(pid)
|
||||
if err != nil {
|
||||
return 0
|
||||
}
|
||||
|
||||
// On Unix, FindProcess always succeeds, so we need to check if it's alive
|
||||
if err := process.Signal(syscall.Signal(0)); err != nil {
|
||||
// Process is not running
|
||||
_ = os.Remove(pidFile)
|
||||
return 0
|
||||
}
|
||||
|
||||
return pid
|
||||
}
|
||||
|
||||
// StopServerByPID stops a dolt sql-server by its PID
|
||||
func StopServerByPID(pid int) error {
|
||||
process, err := os.FindProcess(pid)
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
|
||||
// Try graceful shutdown first
|
||||
if err := process.Signal(syscall.SIGTERM); err != nil {
|
||||
if !strings.Contains(err.Error(), "process already finished") {
|
||||
return err
|
||||
}
|
||||
return nil
|
||||
}
|
||||
|
||||
// Wait for graceful shutdown
|
||||
done := make(chan struct{})
|
||||
go func() {
|
||||
_, _ = process.Wait()
|
||||
close(done)
|
||||
}()
|
||||
|
||||
select {
|
||||
case <-done:
|
||||
return nil
|
||||
case <-time.After(ServerStopTimeout):
|
||||
// Force kill
|
||||
return process.Kill()
|
||||
}
|
||||
}
|
||||
Reference in New Issue
Block a user