wip

go.mod

@@ -10,6 +10,7 @@ require (
 	github.com/klauspost/pgzip v1.2.6
 	github.com/pelletier/go-toml/v2 v2.2.2
 	github.com/qdm12/dns/v2 v2.0.0-rc6
+	github.com/qdm12/goservices v0.1.0
 	github.com/qdm12/gosettings v0.4.2
 	github.com/qdm12/goshutdown v0.3.0
 	github.com/qdm12/gosplash v0.2.0

go.sum

@@ -62,6 +62,8 @@ github.com/prometheus/procfs v0.10.1 h1:kYK1Va/YMlutzCGazswoHKo//tZVlFpKYh+Pymzi
 github.com/prometheus/procfs v0.10.1/go.mod h1:nwNm2aOCAYw8uTR/9bWRREkZFxAUcWzPHWJq+XBB/FM=
 github.com/qdm12/dns/v2 v2.0.0-rc6 h1:h5KpuqZ3IMoSbz2a0OkHzIVc9/jk2vuIm9RoKJuaI78=
 github.com/qdm12/dns/v2 v2.0.0-rc6/go.mod h1:Oh34IJIG55BgHoACOf+cgZCgDiFuiJZ6r6gQW58FN+k=
+github.com/qdm12/goservices v0.1.0 h1:9sODefm/yuIGS7ynCkEnNlMTAYn9GzPhtcK4F69JWvc=
+github.com/qdm12/goservices v0.1.0/go.mod h1:/JOFsAnHFiSjyoXxa5FlfX903h20K5u/3rLzCjYVMck=
 github.com/qdm12/gosettings v0.4.2 h1:Gb39NScPr7OQV+oy0o1OD7A121udITDJuUGa7ljDF58=
 github.com/qdm12/gosettings v0.4.2/go.mod h1:CPrt2YC4UsURTrslmhxocVhMCW03lIrqdH2hzIf5prg=
 github.com/qdm12/goshutdown v0.3.0 h1:pqBpJkdwlZlfTEx4QHtS8u8CXx6pG0fVo6S1N0MpSEM=

internal/configuration/settings/shadowsocks.go

@@ -22,7 +22,7 @@ func (s Shadowsocks) validate() (err error) {
 	return s.Settings.Validate()
 }
 
-func (s *Shadowsocks) copy() (copied Shadowsocks) {
+func (s *Shadowsocks) Copy() (copied Shadowsocks) {
 	return Shadowsocks{
 		Enabled:  gosettings.CopyPointer(s.Enabled),
 		Settings: s.Settings.Copy(),
@@ -32,7 +32,7 @@ func (s *Shadowsocks) copy() (copied Shadowsocks) {
 // overrideWith overrides fields of the receiver
 // settings object with any field set in the other
 // settings.
-func (s *Shadowsocks) overrideWith(other Shadowsocks) {
+func (s *Shadowsocks) OverrideWith(other Shadowsocks) {
 	s.Enabled = gosettings.OverrideWithPointer(s.Enabled, other.Enabled)
 	s.Settings.OverrideWith(other.Settings)
 }

internal/httpproxy/run.go

@@ -18,15 +18,22 @@ func (l *Loop) Run(ctx context.Context, done chan<- struct{}) {
 	}
 
 	for ctx.Err() == nil {
-		runCtx, runCancel := context.WithCancel(ctx)
-
 		settings := l.state.GetSettings()
-		server := New(runCtx, settings.ListeningAddress, l.logger,
+		server, err := New(settings.ListeningAddress, l.logger,
 			*settings.Stealth, *settings.Log, *settings.User,
 			*settings.Password, settings.ReadHeaderTimeout, settings.ReadTimeout)
+		if err != nil {
+			l.statusManager.SetStatus(constants.Crashed)
+			l.logAndWait(ctx, err)
+			continue
+		}
 
-		errorCh := make(chan error)
+		errorCh, err := server.Start(ctx)
-		go server.Run(runCtx, errorCh)
+		if err != nil {
+			l.statusManager.SetStatus(constants.Crashed)
+			l.logAndWait(ctx, err)
+			continue
+		}
 
 		// TODO stable timer, check Shadowsocks
 		if l.userTrigger {
@@ -41,31 +48,23 @@ func (l *Loop) Run(ctx context.Context, done chan<- struct{}) {
 		for stayHere {
 			select {
 			case <-ctx.Done():
-				runCancel()
+				_ = server.Stop()
-				<-errorCh
-				close(errorCh)
 				return
 			case <-l.start:
 				l.userTrigger = true
 				l.logger.Info("starting")
-				runCancel()
+				_ = server.Stop()
-				<-errorCh
-				close(errorCh)
 				stayHere = false
 			case <-l.stop:
 				l.userTrigger = true
 				l.logger.Info("stopping")
-				runCancel()
+				_ = server.Stop()
-				<-errorCh
-				// Do not close errorCh or this for loop won't work
 				l.stopped <- struct{}{}
 			case err := <-errorCh:
-				close(errorCh)
 				l.statusManager.SetStatus(constants.Crashed)
 				l.logAndWait(ctx, err)
 				stayHere = false
 			}
 		}
-		runCancel() // repetition for linter only
 	}
 }

internal/httpproxy/server.go

@@ -2,57 +2,81 @@ package httpproxy
 
 import (
 	"context"
-	"net/http"
+	"fmt"
 	"sync"
 	"time"
+
+	"github.com/qdm12/goservices"
+	"github.com/qdm12/goservices/httpserver"
 )
 
 type Server struct {
-	address           string
+	httpServer    *httpserver.Server
-	handler           http.Handler
+	handlerCtx    context.Context //nolint:containedctx
-	logger            infoErrorer
+	handlerCancel context.CancelFunc
-	internalWG        *sync.WaitGroup
+	handlerWg     *sync.WaitGroup
-	readHeaderTimeout time.Duration
+
-	readTimeout       time.Duration
+	// Server settings
+	httpServerSettings httpserver.Settings
+
+	// Handler settings
+	logger   Logger
+	stealth  bool
+	verbose  bool
+	username string
+	password string
 }
 
-func New(ctx context.Context, address string, logger Logger,
+func ptrTo[T any](x T) *T { return &x }
+
+func New(address string, logger Logger,
 	stealth, verbose bool, username, password string,
 	readHeaderTimeout, readTimeout time.Duration,
-) *Server {
+) (server *Server, err error) {
-	wg := &sync.WaitGroup{}
 	return &Server{
-		address:           address,
+		handlerWg: &sync.WaitGroup{},
-		handler:           newHandler(ctx, wg, logger, stealth, verbose, username, password),
+		httpServerSettings: httpserver.Settings{
-		logger:            logger,
+			// Handler is set when calling Start and reset when Stop is called
-		internalWG:        wg,
+			Handler:           nil,
-		readHeaderTimeout: readHeaderTimeout,
+			Name:              ptrTo("proxy"),
-		readTimeout:       readTimeout,
+			Address:           ptrTo(address),
-	}
+			ReadTimeout:       readTimeout,
+			ReadHeaderTimeout: readHeaderTimeout,
+			Logger:            logger,
+		},
+		logger:   logger,
+		stealth:  stealth,
+		verbose:  verbose,
+		username: username,
+		password: password,
+	}, nil
 }
 
-func (s *Server) Run(ctx context.Context, errorCh chan<- error) {
+func (s *Server) Start(ctx context.Context) (
-	server := http.Server{
+	runError <-chan error, err error,
-		Addr:              s.address,
+) {
-		Handler:           s.handler,
+	if s.httpServer != nil {
-		ReadHeaderTimeout: s.readHeaderTimeout,
+		return nil, fmt.Errorf("%w", goservices.ErrAlreadyStarted)
-		ReadTimeout:       s.readTimeout,
 	}
-	go func() {
+
-		<-ctx.Done()
+	s.handlerCtx, s.handlerCancel = context.WithCancel(context.Background())
-		const shutdownGraceDuration = 100 * time.Millisecond
+	s.httpServerSettings.Handler = newHandler(s.handlerCtx, s.handlerWg,
-		shutdownCtx, cancel := context.WithTimeout(context.Background(), shutdownGraceDuration)
+		s.logger, s.stealth, s.verbose, s.username, s.password)
-		defer cancel()
+	s.httpServer, err = httpserver.New(s.httpServerSettings)
-		if err := server.Shutdown(shutdownCtx); err != nil {
+	if err != nil {
-			s.logger.Error("failed shutting down: " + err.Error())
+		return nil, fmt.Errorf("creating http server: %w", err)
-		}
-	}()
-	s.logger.Info("listening on " + s.address)
-	err := server.ListenAndServe()
-	s.internalWG.Wait()
-	if err != nil && ctx.Err() == nil {
-		errorCh <- err
-	} else {
-		errorCh <- nil
 	}
+
+	return s.httpServer.Start(ctx)
+}
+
+func (s *Server) Stop() (err error) {
+	if s.httpServer == nil {
+		return fmt.Errorf("%w", goservices.ErrAlreadyStopped)
+	}
+	s.handlerCancel()
+	err = s.httpServer.Stop()
+	s.handlerWg.Wait()
+	s.httpServer = nil // signal the server is down
+	return err
 }

internal/shadowsocks/loop.go

@@ -2,13 +2,11 @@ package shadowsocks
 
 import (
 	"context"
-	"sync"
 	"time"
 
 	"github.com/qdm12/gluetun/internal/configuration/settings"
 	"github.com/qdm12/gluetun/internal/constants"
 	"github.com/qdm12/gluetun/internal/models"
-	shadowsockslib "github.com/qdm12/ss-server/pkg/tcpudp"
 )
 
 type Loop struct {
@@ -16,11 +14,110 @@ type Loop struct {
 	// Other objects
 	logger Logger
 	// Internal channels and locks
-	loopLock      sync.Mutex
+	refreshing  bool
-	running       chan models.LoopStatus
+	refresh     chan struct{}
-	stop, stopped chan struct{}
+	changed     chan models.LoopStatus
-	start         chan struct{}
+	backoffTime time.Duration
-	backoffTime   time.Duration
+
+	runCancel context.CancelFunc
+	runDone   <-chan struct{}
+}
+
+const defaultBackoffTime = 10 * time.Second
+
+func NewLoop(settings settings.Shadowsocks, logger Logger) *Loop {
+	return &Loop{
+		state: state{
+			status:   constants.Stopped,
+			settings: settings,
+		},
+		logger:      logger,
+		refresh:     make(chan struct{}, 1), // capacity of 1 to handle crash auto-restart
+		changed:     make(chan models.LoopStatus),
+		backoffTime: defaultBackoffTime,
+	}
+}
+
+func (l *Loop) Start(ctx context.Context) (runError <-chan error, err error) {
+	runCtx, runCancel := context.WithCancel(context.Background())
+	l.runCancel = runCancel
+	ready := make(chan struct{})
+	done := make(chan struct{})
+	l.runDone = done
+
+	go l.run(runCtx, ready, done)
+
+	<-ready
+
+	return nil, nil //nolint:nilnil
+}
+
+func (l *Loop) run(ctx context.Context, ready, done chan<- struct{}) {
+	defer close(done)
+	close(ready)
+
+	for ctx.Err() == nil {
+		// What if update and crash at the same time ish?
+		settings := l.GetSettings()
+
+		var service *service
+		var runError <-chan error
+		var err error
+		if *settings.Enabled {
+			service = newService(settings.Settings, l.logger)
+			runError, err = service.Start(ctx)
+			if err != nil {
+				runErrorCh := make(chan error, 1)
+				runError = runErrorCh
+				runErrorCh <- err
+			} else if l.refreshing {
+				l.changed <- constants.Running
+			} else { // auto-restart due to crash
+				l.state.setStatusWithLock(constants.Running)
+				l.backoffTime = defaultBackoffTime
+			}
+		} else {
+			if l.refreshing {
+				l.changed <- constants.Stopped
+			} else { // auto-restart due to crash
+				l.state.setStatusWithLock(constants.Stopped)
+				l.backoffTime = defaultBackoffTime
+			}
+		}
+		l.refreshing = false
+
+		select {
+		case <-l.refresh:
+			l.refreshing = true
+			if service != nil {
+				err = service.Stop()
+				if err != nil {
+					l.logger.Error("stopping service: " + err.Error())
+				}
+			}
+		case err = <-runError:
+			if l.refreshing {
+				l.changed <- constants.Crashed
+			} else {
+				l.state.setStatusWithLock(constants.Crashed)
+			}
+			l.logAndWait(ctx, err)
+		case <-ctx.Done():
+			if service != nil {
+				err = service.Stop()
+				if err != nil {
+					l.logger.Error("stopping service: " + err.Error())
+				}
+			}
+			return
+		}
+	}
+}
+
+func (l *Loop) Stop() (err error) {
+	l.runCancel()
+	<-l.runDone
+	return nil
 }
 
 func (l *Loop) logAndWait(ctx context.Context, err error) {
@@ -33,113 +130,7 @@ func (l *Loop) logAndWait(ctx context.Context, err error) {
 	select {
 	case <-timer.C:
 	case <-ctx.Done():
-		if !timer.Stop() {
+		_ = timer.Stop()
-			<-timer.C
+	case <-l.refresh: // user-triggered refresh
-		}
-	}
-}
-
-const defaultBackoffTime = 10 * time.Second
-
-func NewLoop(settings settings.Shadowsocks, logger Logger) *Loop {
-	return &Loop{
-		state: state{
-			status:   constants.Stopped,
-			settings: settings,
-		},
-		logger:      logger,
-		start:       make(chan struct{}),
-		running:     make(chan models.LoopStatus),
-		stop:        make(chan struct{}),
-		stopped:     make(chan struct{}),
-		backoffTime: defaultBackoffTime,
-	}
-}
-
-func (l *Loop) Run(ctx context.Context, done chan<- struct{}) {
-	defer close(done)
-
-	crashed := false
-
-	if *l.GetSettings().Enabled {
-		go func() {
-			_, _ = l.SetStatus(ctx, constants.Running)
-		}()
-	}
-
-	select {
-	case <-l.start:
-	case <-ctx.Done():
-		return
-	}
-
-	for ctx.Err() == nil {
-		settings := l.GetSettings()
-		server, err := shadowsockslib.NewServer(settings.Settings, l.logger)
-		if err != nil {
-			crashed = true
-			l.logAndWait(ctx, err)
-			continue
-		}
-
-		shadowsocksCtx, shadowsocksCancel := context.WithCancel(ctx)
-
-		waitError := make(chan error)
-		go func() {
-			waitError <- server.Listen(shadowsocksCtx)
-		}()
-		if err != nil {
-			crashed = true
-			shadowsocksCancel()
-			l.logAndWait(ctx, err)
-			continue
-		}
-
-		isStableTimer := time.NewTimer(time.Second)
-
-		stayHere := true
-		for stayHere {
-			select {
-			case <-ctx.Done():
-				shadowsocksCancel()
-				<-waitError
-				close(waitError)
-				return
-			case <-isStableTimer.C:
-				if !crashed {
-					l.running <- constants.Running
-					crashed = false
-				} else {
-					l.backoffTime = defaultBackoffTime
-					l.state.setStatusWithLock(constants.Running)
-				}
-			case <-l.start:
-				l.logger.Info("starting")
-				shadowsocksCancel()
-				<-waitError
-				close(waitError)
-				stayHere = false
-			case <-l.stop:
-				l.logger.Info("stopping")
-				shadowsocksCancel()
-				<-waitError
-				close(waitError)
-				l.stopped <- struct{}{}
-			case err := <-waitError: // unexpected error
-				shadowsocksCancel()
-				close(waitError)
-				if ctx.Err() != nil {
-					return
-				}
-				l.state.setStatusWithLock(constants.Crashed)
-				l.logAndWait(ctx, err)
-				crashed = true
-				stayHere = false
-			}
-		}
-		shadowsocksCancel() // repetition for linter only
-		if !isStableTimer.Stop() {
-			<-isStableTimer.C
-		}
 	}
 }

internal/shadowsocks/noop.go

@@ -0,0 +1,14 @@
+package shadowsocks
+
+import "context"
+
+type noopService struct{}
+
+func (n *noopService) Start(_ context.Context) (
+	runError <-chan error, err error) {
+	return nil, nil
+}
+
+func (n *noopService) Stop() (err error) {
+	return nil
+}

internal/shadowsocks/service.go

@@ -0,0 +1,67 @@
+package shadowsocks
+
+import (
+	"context"
+	"fmt"
+	"time"
+
+	"github.com/qdm12/ss-server/pkg/tcpudp"
+)
+
+type service struct {
+	// Injected settings
+	settings tcpudp.Settings
+	logger   Logger
+	// Internal fields
+	cancel context.CancelFunc
+	done   <-chan struct{}
+}
+
+func newService(settings tcpudp.Settings,
+	logger Logger) *service {
+	return &service{
+		settings: settings,
+		logger:   logger,
+	}
+}
+
+func (s *service) Start(ctx context.Context) (runError <-chan error, err error) {
+	server, err := tcpudp.NewServer(s.settings, s.logger)
+	if err != nil {
+		return nil, fmt.Errorf("creating server: %w", err)
+	}
+
+	shadowsocksCtx, shadowsocksCancel := context.WithCancel(context.Background())
+	s.cancel = shadowsocksCancel
+	runErrorCh := make(chan error)
+	done := make(chan struct{})
+	s.done = done
+	go func() {
+		defer close(done)
+		err = server.Listen(shadowsocksCtx)
+		if shadowsocksCtx.Err() == nil {
+			runErrorCh <- fmt.Errorf("listening: %w", err)
+		}
+	}()
+
+	const minStabilityTime = 100 * time.Millisecond
+	isStableTimer := time.NewTimer(minStabilityTime)
+	select {
+	case <-isStableTimer.C:
+	case err = <-runErrorCh:
+		return nil, fmt.Errorf("server became unstable within %s: %w",
+			minStabilityTime, err)
+	case <-ctx.Done():
+		shadowsocksCancel()
+		<-done
+		return nil, ctx.Err()
+	}
+
+	return runErrorCh, nil
+}
+
+func (s *service) Stop() (err error) {
+	s.cancel()
+	<-s.done
+	return nil
+}

internal/shadowsocks/state.go

@@ -1,14 +1,10 @@
 package shadowsocks
 
 import (
-	"context"
-	"errors"
-	"fmt"
 	"reflect"
 	"sync"
 
 	"github.com/qdm12/gluetun/internal/configuration/settings"
-	"github.com/qdm12/gluetun/internal/constants"
 	"github.com/qdm12/gluetun/internal/models"
 )
 
@@ -25,95 +21,34 @@ func (s *state) setStatusWithLock(status models.LoopStatus) {
 	s.status = status
 }
 
+// GetStatus returns the status of the loop for informative purposes.
+// In no case it should be used programmatically to avoid any
+// TOCTOU race conditions.
 func (l *Loop) GetStatus() (status models.LoopStatus) {
 	l.state.statusMu.RLock()
 	defer l.state.statusMu.RUnlock()
 	return l.state.status
 }
 
-var ErrInvalidStatus = errors.New("invalid status")
-
-func (l *Loop) SetStatus(ctx context.Context, status models.LoopStatus) (
-	outcome string, err error,
-) {
-	l.state.statusMu.Lock()
-	defer l.state.statusMu.Unlock()
-	existingStatus := l.state.status
-
-	switch status {
-	case constants.Running:
-		switch existingStatus {
-		case constants.Starting, constants.Running, constants.Stopping, constants.Crashed:
-			return fmt.Sprintf("already %s", existingStatus), nil
-		}
-		l.loopLock.Lock()
-		defer l.loopLock.Unlock()
-		l.state.status = constants.Starting
-		l.state.statusMu.Unlock()
-		l.start <- struct{}{}
-
-		newStatus := constants.Starting // for canceled context
-		select {
-		case <-ctx.Done():
-		case newStatus = <-l.running:
-		}
-		l.state.statusMu.Lock()
-		l.state.status = newStatus
-		return newStatus.String(), nil
-	case constants.Stopped:
-		switch existingStatus {
-		case constants.Stopped, constants.Stopping, constants.Starting, constants.Crashed:
-			return fmt.Sprintf("already %s", existingStatus), nil
-		}
-		l.loopLock.Lock()
-		defer l.loopLock.Unlock()
-		l.state.status = constants.Stopping
-		l.state.statusMu.Unlock()
-		l.stop <- struct{}{}
-		newStatus := constants.Stopping // for canceled context
-		select {
-		case <-ctx.Done():
-		case <-l.stopped:
-			newStatus = constants.Stopped
-		}
-		l.state.statusMu.Lock()
-		l.state.status = newStatus
-		return status.String(), nil
-	default:
-		return "", fmt.Errorf("%w: %s: it can only be one of: %s, %s",
-			ErrInvalidStatus, status, constants.Running, constants.Stopped)
-	}
-}
-
 func (l *Loop) GetSettings() (settings settings.Shadowsocks) {
 	l.state.settingsMu.RLock()
 	defer l.state.settingsMu.RUnlock()
 	return l.state.settings
 }
 
-func (l *Loop) SetSettings(ctx context.Context, settings settings.Shadowsocks) (
+func (l *Loop) UpdateSettings(updateSettings settings.Shadowsocks) (outcome string) {
-	outcome string,
-) {
 	l.state.settingsMu.Lock()
-	settingsUnchanged := reflect.DeepEqual(settings, l.state.settings)
+	previousSettings := l.state.settings.Copy()
+	l.state.settings.OverrideWith(updateSettings)
+	settingsUnchanged := reflect.DeepEqual(previousSettings, l.state.settings)
+	l.state.settingsMu.Unlock()
 	if settingsUnchanged {
-		l.state.settingsMu.Unlock()
 		return "settings left unchanged"
 	}
-	newEnabled := *settings.Enabled
+	l.refresh <- struct{}{}
-	previousEnabled := *l.state.settings.Enabled
+	newStatus := <-l.changed
-	l.state.settings = settings
+	l.state.statusMu.Lock()
-	l.state.settingsMu.Unlock()
+	l.state.status = newStatus
-	// Either restart or set changed status
+	l.state.statusMu.Unlock()
-	switch {
+	return "settings updated (service " + newStatus.String() + ")"
-	case !newEnabled && !previousEnabled:
-	case newEnabled && previousEnabled:
-		_, _ = l.SetStatus(ctx, constants.Stopped)
-		_, _ = l.SetStatus(ctx, constants.Running)
-	case newEnabled && !previousEnabled:
-		_, _ = l.SetStatus(ctx, constants.Running)
-	case !newEnabled && previousEnabled:
-		_, _ = l.SetStatus(ctx, constants.Stopped)
-	}
-	return "settings updated"
 }