Fix panic conditions

Due to the order of the previous logic in ScanReader, an error not caused by EOF would effectively get ignored since an error will always be returned with `isPrefix` equal to false, thus triggering the first break, and error checking is not performed beyond that point.

Thus, canceling an installation process for a server while this process was running would hang the routine and cause the loop to run endlessly, even with a canceled context.

Makefile

@@ -6,7 +6,7 @@ build:
 
 debug:
 	go build -ldflags="-X github.com/pterodactyl/wings/system.Version=$(GIT_HEAD)"
-	sudo ./wings --debug --ignore-certificate-errors --config config.yml --pprof
+	sudo ./wings --debug --ignore-certificate-errors --config config.yml --pprof --pprof-block-rate 1
 
 # Runs a remotly debuggable session for Wings allowing an IDE to connect and target
 # different breakpoints.

cmd/root.go

@@ -11,6 +11,7 @@ import (
 	"os"
 	"path"
 	"path/filepath"
+	"runtime"
 	"strconv"
 	"strings"
 	"time"
@@ -76,6 +77,7 @@ func init() {
 
 	// Flags specifically used when running the API.
 	rootCommand.Flags().Bool("pprof", false, "if the pprof profiler should be enabled. The profiler will bind to localhost:6060 by default")
+	rootCommand.Flags().Int("pprof-block-rate", 0, "enables block profile support, may have performance impacts")
 	rootCommand.Flags().Int("pprof-port", 6060, "If provided with --pprof, the port it will run on")
 	rootCommand.Flags().Bool("auto-tls", false, "pass in order to have wings generate and manage it's own SSL certificates using Let's Encrypt")
 	rootCommand.Flags().String("tls-hostname", "", "required with --auto-tls, the FQDN for the generated SSL certificate")
@@ -309,6 +311,12 @@ func rootCmdRun(cmd *cobra.Command, _ []string) {
 
 	profile, _ := cmd.Flags().GetBool("pprof")
 	if profile {
+		if r, _ := cmd.Flags().GetInt("pprof-block-rate"); r > 0 {
+			runtime.SetBlockProfileRate(r)
+		}
+		// Catch at least 1% of mutex contention issues.
+		runtime.SetMutexProfileFraction(100)
+
 		profilePort, _ := cmd.Flags().GetInt("pprof-port")
 		go func() {
 			http.ListenAndServe(fmt.Sprintf("localhost:%d", profilePort), nil)

environment/docker/api.go

@@ -73,6 +73,9 @@ func (e *Environment) ContainerInspect(ctx context.Context) (types.ContainerJSON
 
 	res, err := e.client.HTTPClient().Do(req)
 	if err != nil {
+		if res == nil {
+			return st, errdefs.Unknown(err)
+		}
 		return st, errdefs.FromStatusCode(err, res.StatusCode)
 	}
 

environment/docker/power.go

@@ -138,9 +138,7 @@ func (e *Environment) Start(ctx context.Context) error {
 // You most likely want to be using WaitForStop() rather than this function,
 // since this will return as soon as the command is sent, rather than waiting
 // for the process to be completed stopped.
-//
+func (e *Environment) Stop(ctx context.Context) error {
-// TODO: pass context through from the server instance.
-func (e *Environment) Stop() error {
 	e.mu.RLock()
 	s := e.meta.Stop
 	e.mu.RUnlock()
@@ -164,7 +162,7 @@ func (e *Environment) Stop() error {
 		case "SIGTERM":
 			signal = syscall.SIGTERM
 		}
-		return e.Terminate(signal)
+		return e.Terminate(ctx, signal)
 	}
 
 	// If the process is already offline don't switch it back to stopping. Just leave it how
@@ -179,8 +177,10 @@ func (e *Environment) Stop() error {
 		return e.SendCommand(s.Value)
 	}
 
-	t := time.Second * 30
+	// Allow the stop action to run for however long it takes, similar to executing a command
-	if err := e.client.ContainerStop(context.Background(), e.Id, &t); err != nil {
+	// and using a different logic pathway to wait for the container to stop successfully.
+	t := time.Duration(-1)
+	if err := e.client.ContainerStop(ctx, e.Id, &t); err != nil {
 		// If the container does not exist just mark the process as stopped and return without
 		// an error.
 		if client.IsErrNotFound(err) {
@@ -198,45 +198,66 @@ func (e *Environment) Stop() error {
 // command. If the server does not stop after seconds have passed, an error will
 // be returned, or the instance will be terminated forcefully depending on the
 // value of the second argument.
-func (e *Environment) WaitForStop(seconds uint, terminate bool) error {
+//
-	if err := e.Stop(); err != nil {
+// Calls to Environment.Terminate() in this function use the context passed
-		return err
+// through since we don't want to prevent termination of the server instance
+// just because the context.WithTimeout() has expired.
+func (e *Environment) WaitForStop(ctx context.Context, duration time.Duration, terminate bool) error {
+	tctx, cancel := context.WithTimeout(context.Background(), duration)
+	defer cancel()
+
+	// If the parent context is canceled, abort the timed context for termination.
+	go func() {
+		select {
+		case <-ctx.Done():
+			cancel()
+		case <-tctx.Done():
+			// When the timed context is canceled, terminate this routine since we no longer
+			// need to worry about the parent routine being canceled.
+			break
+		}
+	}()
+
+	doTermination := func (s string) error {
+		e.log().WithField("step", s).WithField("duration", duration).Warn("container stop did not complete in time, terminating process...")
+		return e.Terminate(ctx, os.Kill)
 	}
 
-	ctx, cancel := context.WithTimeout(context.Background(), time.Duration(seconds)*time.Second)
+	// We pass through the timed context for this stop action so that if one of the
-	defer cancel()
+	// internal docker calls fails to ever finish before we've exhausted the time limit
+	// the resources get cleaned up, and the exection is stopped.
+	if err := e.Stop(tctx); err != nil {
+		if terminate && errors.Is(err, context.DeadlineExceeded) {
+			return doTermination("stop")
+		}
+		return err
+	}
 
 	// Block the return of this function until the container as been marked as no
 	// longer running. If this wait does not end by the time seconds have passed,
 	// attempt to terminate the container, or return an error.
-	ok, errChan := e.client.ContainerWait(ctx, e.Id, container.WaitConditionNotRunning)
+	ok, errChan := e.client.ContainerWait(tctx, e.Id, container.WaitConditionNotRunning)
 	select {
 	case <-ctx.Done():
-		if ctxErr := ctx.Err(); ctxErr != nil {
+		if err := ctx.Err(); err != nil {
 			if terminate {
-				log.WithField("container_id", e.Id).Info("server did not stop in time, executing process termination")
+				return doTermination("parent-context")
-
-				return e.Terminate(os.Kill)
 			}
-
+			return err
-			return ctxErr
 		}
 	case err := <-errChan:
 		// If the error stems from the container not existing there is no point in wasting
 		// CPU time to then try and terminate it.
-		if err != nil && !client.IsErrNotFound(err) {
+		if err == nil || client.IsErrNotFound(err) {
-			if terminate {
+			return nil
-				l := log.WithField("container_id", e.Id)
-				if errors.Is(err, context.DeadlineExceeded) {
-					l.Warn("deadline exceeded for container stop; terminating process")
-				} else {
-					l.WithField("error", err).Warn("error while waiting for container stop; terminating process")
 		}
-
+		if terminate {
-				return e.Terminate(os.Kill)
+			if !errors.Is(err, context.DeadlineExceeded) {
+				e.log().WithField("error", err).Warn("error while waiting for container stop; terminating process")
+			}
+			return doTermination("wait")
 		}
 		return errors.WrapIf(err, "environment/docker: error waiting on container to enter \"not-running\" state")
-		}
 	case <-ok:
 	}
 
@@ -244,8 +265,8 @@ func (e *Environment) WaitForStop(seconds uint, terminate bool) error {
 }
 
 // Terminate forcefully terminates the container using the signal provided.
-func (e *Environment) Terminate(signal os.Signal) error {
+func (e *Environment) Terminate(ctx context.Context, signal os.Signal) error {
-	c, err := e.ContainerInspect(context.Background())
+	c, err := e.ContainerInspect(ctx)
 	if err != nil {
 		// Treat missing containers as an okay error state, means it is obviously
 		// already terminated at this point.
@@ -270,7 +291,7 @@ func (e *Environment) Terminate(signal os.Signal) error {
 	// We set it to stopping than offline to prevent crash detection from being triggered.
 	e.SetState(environment.ProcessStoppingState)
 	sig := strings.TrimSuffix(strings.TrimPrefix(signal.String(), "signal "), "ed")
-	if err := e.client.ContainerKill(context.Background(), e.Id, sig); err != nil && !client.IsErrNotFound(err) {
+	if err := e.client.ContainerKill(ctx, e.Id, sig); err != nil && !client.IsErrNotFound(err) {
 		return errors.WithStack(err)
 	}
 	e.SetState(environment.ProcessOfflineState)

environment/environment.go

@@ -3,6 +3,7 @@ package environment
 import (
 	"context"
 	"os"
+	"time"
 
 	"github.com/pterodactyl/wings/events"
 )
@@ -58,18 +59,20 @@ type ProcessEnvironment interface {
 	// can be started an error should be returned.
 	Start(ctx context.Context) error
 
-	// Stops a server instance. If the server is already stopped an error should
+	// Stop stops a server instance. If the server is already stopped an error will
-	// not be returned.
+	// not be returned, this function will act as a no-op.
-	Stop() error
+	Stop(ctx context.Context) error
 
-	// Waits for a server instance to stop gracefully. If the server is still detected
+	// WaitForStop waits for a server instance to stop gracefully. If the server is
-	// as running after seconds, an error will be returned, or the server will be terminated
+	// still detected as running after "duration", an error will be returned, or the server
-	// depending on the value of the second argument.
+	// will be terminated depending on the value of the second argument. If the context
-	WaitForStop(seconds uint, terminate bool) error
+	// provided is canceled the underlying wait conditions will be stopped and the
+	// entire loop will be ended (potentially without stopping or terminating).
+	WaitForStop(ctx context.Context, duration time.Duration, terminate bool) error
 
-	// Terminates a running server instance using the provided signal. If the server
+	// Terminate stops a running server instance using the provided signal. This function
-	// is not running no error should be returned.
+	// is a no-op if the server is already stopped.
-	Terminate(signal os.Signal) error
+	Terminate(ctx context.Context, signal os.Signal) error
 
 	// Destroys the environment removing any containers that were created (in Docker
 	// environments at least).

events/events.go

@@ -2,10 +2,11 @@ package events
 
 import (
 	"strings"
-	"sync"
-)
 
-type Listener chan Event
+	"emperror.dev/errors"
+	"github.com/goccy/go-json"
+	"github.com/pterodactyl/wings/system"
+)
 
 // Event represents an Event sent over a Bus.
 type Event struct {
@@ -15,137 +16,55 @@ type Event struct {
 
 // Bus represents an Event Bus.
 type Bus struct {
-	listenersMx sync.Mutex
+	*system.SinkPool
-	listeners   map[string][]Listener
 }
 
-// NewBus returns a new empty Event Bus.
+// NewBus returns a new empty Bus. This is simply a nicer wrapper around the
+// system.SinkPool implementation that allows for more simplistic usage within
+// the codebase.
+//
+// All of the events emitted out of this bus are byte slices that can be decoded
+// back into an events.Event interface.
 func NewBus() *Bus {
 	return &Bus{
-		listeners: make(map[string][]Listener),
+		system.NewSinkPool(),
-	}
-}
-
-// Off unregisters a listener from the specified topics on the Bus.
-func (b *Bus) Off(listener Listener, topics ...string) {
-	b.listenersMx.Lock()
-	defer b.listenersMx.Unlock()
-
-	var closed bool
-
-	for _, topic := range topics {
-		ok := b.off(topic, listener)
-		if !closed && ok {
-			close(listener)
-			closed = true
-		}
-	}
-}
-
-func (b *Bus) off(topic string, listener Listener) bool {
-	listeners, ok := b.listeners[topic]
-	if !ok {
-		return false
-	}
-	for i, l := range listeners {
-		if l != listener {
-			continue
-		}
-
-		listeners = append(listeners[:i], listeners[i+1:]...)
-		b.listeners[topic] = listeners
-		return true
-	}
-	return false
-}
-
-// On registers a listener to the specified topics on the Bus.
-func (b *Bus) On(listener Listener, topics ...string) {
-	b.listenersMx.Lock()
-	defer b.listenersMx.Unlock()
-
-	for _, topic := range topics {
-		b.on(topic, listener)
-	}
-}
-
-func (b *Bus) on(topic string, listener Listener) {
-	listeners, ok := b.listeners[topic]
-	if !ok {
-		b.listeners[topic] = []Listener{listener}
-	} else {
-		b.listeners[topic] = append(listeners, listener)
 	}
 }
 
 // Publish publishes a message to the Bus.
 func (b *Bus) Publish(topic string, data interface{}) {
-	// Some of our topics for the socket support passing a more specific namespace,
+	// Some of our actions for the socket support passing a more specific namespace,
 	// such as "backup completed:1234" to indicate which specific backup was completed.
 	//
 	// In these cases, we still need to send the event using the standard listener
 	// name of "backup completed".
 	if strings.Contains(topic, ":") {
 		parts := strings.SplitN(topic, ":", 2)
-
 		if len(parts) == 2 {
 			topic = parts[0]
 		}
 	}
 
-	b.listenersMx.Lock()
+	enc, err := json.Marshal(Event{Topic: topic, Data: data})
-	defer b.listenersMx.Unlock()
+	if err != nil {
-
+		panic(errors.WithStack(err))
-	listeners, ok := b.listeners[topic]
+	}
-	if !ok {
+	b.Push(enc)
-		return
+}
+
+// MustDecode decodes the event byte slice back into an events.Event struct or
+// panics if an error is encountered during this process.
+func MustDecode(data []byte) (e Event) {
+	if err := DecodeTo(data, &e); err != nil {
+		panic(err)
 	}
-	if len(listeners) < 1 {
 	return
 }
 
-	var wg sync.WaitGroup
+// DecodeTo decodes a byte slice of event data into the given interface.
-	event := Event{Topic: topic, Data: data}
+func DecodeTo(data []byte, v interface{}) error {
-	for _, listener := range listeners {
+	if err := json.Unmarshal(data, &v); err != nil {
-		l := listener
+		return errors.Wrap(err, "events: failed to decode byte slice")
-		wg.Add(1)
-		go func(l Listener, event Event) {
-			defer wg.Done()
-			l <- event
-		}(l, event)
 	}
-	wg.Wait()
+	return nil
-}
-
-// Destroy destroys the Event Bus by unregistering and closing all listeners.
-func (b *Bus) Destroy() {
-	b.listenersMx.Lock()
-	defer b.listenersMx.Unlock()
-
-	// Track what listeners have already been closed. Because the same listener
-	// can be listening on multiple topics, we need a way to essentially
-	// "de-duplicate" all the listeners across all the topics.
-	var closed []Listener
-
-	for _, listeners := range b.listeners {
-		for _, listener := range listeners {
-			if contains(closed, listener) {
-				continue
-			}
-
-			close(listener)
-			closed = append(closed, listener)
-		}
-	}
-
-	b.listeners = make(map[string][]Listener)
-}
-
-func contains(closed []Listener, listener Listener) bool {
-	for _, c := range closed {
-		if c == listener {
-			return true
-		}
-	}
-	return false
 }

events/events_test.go

@@ -9,107 +9,34 @@ import (
 
 func TestNewBus(t *testing.T) {
 	g := Goblin(t)
-	bus := NewBus()
+
+	g.Describe("Events", func() {
+		var bus *Bus
+		g.BeforeEach(func() {
+			bus = NewBus()
+		})
 
 		g.Describe("NewBus", func() {
 			g.It("is not nil", func() {
 				g.Assert(bus).IsNotNil("Bus expected to not be nil")
-			g.Assert(bus.listeners).IsNotNil("Bus#listeners expected to not be nil")
 			})
 		})
-}
-
-func TestBus_Off(t *testing.T) {
-	g := Goblin(t)
-
-	const topic = "test"
-
-	g.Describe("Off", func() {
-		g.It("unregisters listener", func() {
-			bus := NewBus()
-
-			g.Assert(bus.listeners[topic]).IsNotNil()
-			g.Assert(len(bus.listeners[topic])).IsZero()
-			listener := make(chan Event)
-			bus.On(listener, topic)
-			g.Assert(len(bus.listeners[topic])).Equal(1, "Listener was not registered")
-
-			bus.Off(listener, topic)
-			g.Assert(len(bus.listeners[topic])).Equal(0, "Topic still has one or more listeners")
-		})
-
-		g.It("unregisters correct listener", func() {
-			bus := NewBus()
-
-			listener := make(chan Event)
-			listener2 := make(chan Event)
-			listener3 := make(chan Event)
-			bus.On(listener, topic)
-			bus.On(listener2, topic)
-			bus.On(listener3, topic)
-			g.Assert(len(bus.listeners[topic])).Equal(3, "Listeners were not registered")
-
-			bus.Off(listener, topic)
-			bus.Off(listener3, topic)
-			g.Assert(len(bus.listeners[topic])).Equal(1, "Expected 1 listener to remain")
-
-			if bus.listeners[topic][0] != listener2 {
-				// A normal Assert does not properly compare channels.
-				g.Fail("wrong listener unregistered")
-			}
-
-			// Cleanup
-			bus.Off(listener2, topic)
-		})
-	})
-}
-
-func TestBus_On(t *testing.T) {
-	g := Goblin(t)
-
-	const topic = "test"
-
-	g.Describe("On", func() {
-		g.It("registers listener", func() {
-			bus := NewBus()
-
-			g.Assert(bus.listeners[topic]).IsNotNil()
-			g.Assert(len(bus.listeners[topic])).IsZero()
-			listener := make(chan Event)
-			bus.On(listener, topic)
-			g.Assert(len(bus.listeners[topic])).Equal(1, "Listener was not registered")
-
-			if bus.listeners[topic][0] != listener {
-				// A normal Assert does not properly compare channels.
-				g.Fail("wrong listener registered")
-			}
-
-			// Cleanup
-			bus.Off(listener, topic)
-		})
-	})
-}
-
-func TestBus_Publish(t *testing.T) {
-	g := Goblin(t)
 
+		g.Describe("Publish", func() {
 			const topic = "test"
 			const message = "this is a test message!"
 
-	g.Describe("Publish", func() {
 			g.It("publishes message", func() {
 				bus := NewBus()
 
-			g.Assert(bus.listeners[topic]).IsNotNil()
+				listener := make(chan []byte)
-			g.Assert(len(bus.listeners[topic])).IsZero()
+				bus.On(listener)
-			listener := make(chan Event)
-			bus.On(listener, topic)
-			g.Assert(len(bus.listeners[topic])).Equal(1, "Listener was not registered")
 
 				done := make(chan struct{}, 1)
 				go func() {
 					select {
-				case m := <-listener:
+					case v := <-listener:
+						m := MustDecode(v)
 						g.Assert(m.Topic).Equal(topic)
 						g.Assert(m.Data).Equal(message)
 					case <-time.After(1 * time.Second):
@@ -121,33 +48,33 @@ func TestBus_Publish(t *testing.T) {
 				<-done
 
 				// Cleanup
-			bus.Off(listener, topic)
+				bus.Off(listener)
 			})
 
 			g.It("publishes message to all listeners", func() {
 				bus := NewBus()
 
-			g.Assert(bus.listeners[topic]).IsNotNil()
+				listener := make(chan []byte)
-			g.Assert(len(bus.listeners[topic])).IsZero()
+				listener2 := make(chan []byte)
-			listener := make(chan Event)
+				listener3 := make(chan []byte)
-			listener2 := make(chan Event)
+				bus.On(listener)
-			listener3 := make(chan Event)
+				bus.On(listener2)
-			bus.On(listener, topic)
+				bus.On(listener3)
-			bus.On(listener2, topic)
-			bus.On(listener3, topic)
-			g.Assert(len(bus.listeners[topic])).Equal(3, "Listener was not registered")
 
 				done := make(chan struct{}, 1)
 				go func() {
 					for i := 0; i < 3; i++ {
 						select {
-					case m := <-listener:
+						case v := <-listener:
+							m := MustDecode(v)
 							g.Assert(m.Topic).Equal(topic)
 							g.Assert(m.Data).Equal(message)
-					case m := <-listener2:
+						case v := <-listener2:
+							m := MustDecode(v)
 							g.Assert(m.Topic).Equal(topic)
 							g.Assert(m.Data).Equal(message)
-					case m := <-listener3:
+						case v := <-listener3:
+							m := MustDecode(v)
 							g.Assert(m.Topic).Equal(topic)
 							g.Assert(m.Data).Equal(message)
 						case <-time.After(1 * time.Second):
@@ -162,9 +89,10 @@ func TestBus_Publish(t *testing.T) {
 				<-done
 
 				// Cleanup
-			bus.Off(listener, topic)
+				bus.Off(listener)
-			bus.Off(listener2, topic)
+				bus.Off(listener2)
-			bus.Off(listener3, topic)
+				bus.Off(listener3)
+			})
 		})
 	})
 }

parser/parser.go

@@ -11,9 +11,9 @@ import (
 	"github.com/apex/log"
 	"github.com/beevik/etree"
 	"github.com/buger/jsonparser"
+	"github.com/goccy/go-json"
 	"github.com/icza/dyno"
 	"github.com/magiconair/properties"
-	"github.com/goccy/go-json"
 	"gopkg.in/ini.v1"
 	"gopkg.in/yaml.v2"
 

router/router_transfer.go

@@ -178,7 +178,7 @@ func postServerArchive(c *gin.Context) {
 
 		// Ensure the server is offline. Sometimes a "No such container" error gets through
 		// which means the server is already stopped. We can ignore that.
-		if err := s.Environment.WaitForStop(60, false); err != nil && !strings.Contains(strings.ToLower(err.Error()), "no such container") {
+		if err := s.Environment.WaitForStop(s.Context(), time.Minute, false); err != nil && !strings.Contains(strings.ToLower(err.Error()), "no such container") {
 			sendTransferLog("Failed to stop server, aborting transfer..")
 			l.WithField("error", err).Error("failed to stop server")
 			return

router/websocket/listeners.go

@@ -7,8 +7,9 @@ import (
 
 	"emperror.dev/errors"
 	"github.com/goccy/go-json"
-
 	"github.com/pterodactyl/wings/events"
+	"github.com/pterodactyl/wings/system"
+
 	"github.com/pterodactyl/wings/server"
 )
 
@@ -88,12 +89,13 @@ func (h *Handler) listenForServerEvents(ctx context.Context) error {
 	ctx, cancel := context.WithCancel(ctx)
 	defer cancel()
 
-	eventChan := make(chan events.Event)
+	eventChan := make(chan []byte)
 	logOutput := make(chan []byte, 8)
 	installOutput := make(chan []byte, 4)
-	h.server.Events().On(eventChan, e...)
+
-	h.server.Sink(server.LogSink).On(logOutput)
+	h.server.Events().On(eventChan) // TODO: make a sinky
-	h.server.Sink(server.InstallSink).On(installOutput)
+	h.server.Sink(system.LogSink).On(logOutput)
+	h.server.Sink(system.InstallSink).On(installOutput)
 
 	onError := func(evt string, err2 error) {
 		h.Logger().WithField("event", evt).WithField("error", err2).Error("failed to send event over server websocket")
@@ -110,19 +112,23 @@ func (h *Handler) listenForServerEvents(ctx context.Context) error {
 		select {
 		case <-ctx.Done():
 			break
-		case e := <-logOutput:
+		case b := <-logOutput:
-			sendErr := h.SendJson(Message{Event: server.ConsoleOutputEvent, Args: []string{string(e)}})
+			sendErr := h.SendJson(Message{Event: server.ConsoleOutputEvent, Args: []string{string(b)}})
 			if sendErr == nil {
 				continue
 			}
 			onError(server.ConsoleOutputEvent, sendErr)
-		case e := <-installOutput:
+		case b := <-installOutput:
-			sendErr := h.SendJson(Message{Event: server.InstallOutputEvent, Args: []string{string(e)}})
+			sendErr := h.SendJson(Message{Event: server.InstallOutputEvent, Args: []string{string(b)}})
 			if sendErr == nil {
 				continue
 			}
 			onError(server.InstallOutputEvent, sendErr)
-		case e := <-eventChan:
+		case b := <-eventChan:
+			var e events.Event
+			if err := events.DecodeTo(b, &e); err != nil {
+				continue
+			}
 			var sendErr error
 			message := Message{Event: e.Topic}
 			if str, ok := e.Data.(string); ok {
@@ -148,9 +154,9 @@ func (h *Handler) listenForServerEvents(ctx context.Context) error {
 	}
 
 	// These functions will automatically close the channel if it hasn't been already.
-	h.server.Events().Off(eventChan, e...)
+	h.server.Events().Off(eventChan)
-	h.server.Sink(server.LogSink).Off(logOutput)
+	h.server.Sink(system.LogSink).Off(logOutput)
-	h.server.Sink(server.InstallSink).Off(installOutput)
+	h.server.Sink(system.InstallSink).Off(installOutput)
 
 	// If the internal context is stopped it is either because the parent context
 	// got canceled or because we ran into an error. If the "err" variable is nil

router/websocket/websocket.go

@@ -11,9 +11,10 @@ import (
 	"emperror.dev/errors"
 	"github.com/apex/log"
 	"github.com/gbrlsnchs/jwt/v3"
+	"github.com/goccy/go-json"
 	"github.com/google/uuid"
 	"github.com/gorilla/websocket"
-	"github.com/goccy/go-json"
+	"github.com/pterodactyl/wings/system"
 
 	"github.com/pterodactyl/wings/config"
 	"github.com/pterodactyl/wings/environment"
@@ -353,7 +354,7 @@ func (h *Handler) HandleInbound(ctx context.Context, m Message) error {
 			}
 
 			err := h.server.HandlePowerAction(action)
-			if errors.Is(err, context.DeadlineExceeded) {
+			if errors.Is(err, system.ErrLockerLocked) {
 				m, _ := h.GetErrorMessage("another power action is currently being processed for this server, please try again later")
 
 				_ = h.SendJson(Message{

server/backup.go

@@ -142,7 +142,7 @@ func (s *Server) RestoreBackup(b backup.BackupInterface, reader io.ReadCloser) (
 	// instance, otherwise you'll likely hit all types of write errors due to the
 	// server being suspended.
 	if s.Environment.State() != environment.ProcessOfflineState {
-		if err = s.Environment.WaitForStop(120, false); err != nil {
+		if err = s.Environment.WaitForStop(s.Context(), time.Minute*2, false); err != nil {
 			if !client.IsErrNotFound(err) {
 				return errors.WrapIf(err, "server/backup: restore: failed to wait for container stop")
 			}

server/config_parser.go

@@ -6,12 +6,14 @@ import (
 	"github.com/gammazero/workerpool"
 )
 
-// Parent function that will update all of the defined configuration files for a server
+// UpdateConfigurationFiles updates all of the defined configuration files for
-// automatically to ensure that they always use the specified values.
+// a server automatically to ensure that they always use the specified values.
 func (s *Server) UpdateConfigurationFiles() {
 	pool := workerpool.New(runtime.NumCPU())
 
+	s.Log().Debug("acquiring process configuration files...")
 	files := s.ProcessConfiguration().ConfigurationFiles
+	s.Log().Debug("acquired process configuration files")
 	for _, cf := range files {
 		f := cf
 
@@ -26,6 +28,8 @@ func (s *Server) UpdateConfigurationFiles() {
 			if err := f.Parse(p, false); err != nil {
 				s.Log().WithField("error", err).Error("failed to parse and update server configuration file")
 			}
+
+			s.Log().WithField("path", f.FileName).Debug("finished processing server configuration file")
 		})
 	}
 

server/events.go

@@ -2,6 +2,7 @@ package server
 
 import (
 	"github.com/pterodactyl/wings/events"
+	"github.com/pterodactyl/wings/system"
 )
 
 // Defines all of the possible output events for a server.
@@ -20,7 +21,7 @@ const (
 	TransferStatusEvent         = "transfer status"
 )
 
-// Returns the server's emitter instance.
+// Events returns the server's emitter instance.
 func (s *Server) Events() *events.Bus {
 	s.emitterLock.Lock()
 	defer s.emitterLock.Unlock()
@@ -31,3 +32,24 @@ func (s *Server) Events() *events.Bus {
 
 	return s.emitter
 }
+
+// Sink returns the instantiated and named sink for a server. If the sink has
+// not been configured yet this function will cause a panic condition.
+func (s *Server) Sink(name system.SinkName) *system.SinkPool {
+	sink, ok := s.sinks[name]
+	if !ok {
+		s.Log().Fatalf("attempt to access nil sink: %s", name)
+	}
+	return sink
+}
+
+// DestroyAllSinks iterates over all of the sinks configured for the server and
+// destroys their instances. Note that this will cause a panic if you attempt
+// to call Server.Sink() again after. This function is only used when a server
+// is being deleted from the system.
+func (s *Server) DestroyAllSinks() {
+	s.Log().Info("destroying all registered sinks for server instance")
+	for _, sink := range s.sinks {
+		sink.Destroy()
+	}
+}

server/install.go

@@ -10,6 +10,7 @@ import (
 	"path/filepath"
 	"strconv"
 	"strings"
+	"time"
 
 	"emperror.dev/errors"
 	"github.com/apex/log"
@@ -17,18 +18,18 @@ import (
 	"github.com/docker/docker/api/types/container"
 	"github.com/docker/docker/api/types/mount"
 	"github.com/docker/docker/client"
-
 	"github.com/pterodactyl/wings/config"
 	"github.com/pterodactyl/wings/environment"
 	"github.com/pterodactyl/wings/remote"
 	"github.com/pterodactyl/wings/system"
 )
 
-// Executes the installation stack for a server process. Bubbles any errors up to the calling
+// Install executes the installation stack for a server process. Bubbles any
-// function which should handle contacting the panel to notify it of the server state.
+// errors up to the calling function which should handle contacting the panel to
+// notify it of the server state.
 //
-// Pass true as the first argument in order to execute a server sync before the process to
+// Pass true as the first argument in order to execute a server sync before the
-// ensure the latest information is used.
+// process to ensure the latest information is used.
 func (s *Server) Install(sync bool) error {
 	if sync {
 		s.Log().Info("syncing server state with remote source before executing installation process")
@@ -79,7 +80,7 @@ func (s *Server) Install(sync bool) error {
 func (s *Server) Reinstall() error {
 	if s.Environment.State() != environment.ProcessOfflineState {
 		s.Log().Debug("waiting for server instance to enter a stopped state")
-		if err := s.Environment.WaitForStop(10, true); err != nil {
+		if err := s.Environment.WaitForStop(s.Context(), time.Second*10, true); err != nil {
 			return err
 		}
 	}
@@ -110,9 +111,7 @@ func (s *Server) internalInstall() error {
 type InstallationProcess struct {
 	Server *Server
 	Script *remote.InstallationScript
-
 	client *client.Client
-	context context.Context
 }
 
 // Generates a new installation process struct that will be used to create containers,
@@ -127,7 +126,6 @@ func NewInstallationProcess(s *Server, script *remote.InstallationScript) (*Inst
 		return nil, err
 	} else {
 		proc.client = c
-		proc.context = s.Context()
 	}
 
 	return proc, nil
@@ -157,7 +155,7 @@ func (s *Server) SetRestoring(state bool) {
 
 // Removes the installer container for the server.
 func (ip *InstallationProcess) RemoveContainer() error {
-	err := ip.client.ContainerRemove(ip.context, ip.Server.ID()+"_installer", types.ContainerRemoveOptions{
+	err := ip.client.ContainerRemove(ip.Server.Context(), ip.Server.ID()+"_installer", types.ContainerRemoveOptions{
 		RemoveVolumes: true,
 		Force:         true,
 	})
@@ -167,11 +165,10 @@ func (ip *InstallationProcess) RemoveContainer() error {
 	return nil
 }
 
-// Runs the installation process, this is done as in a background thread. This will configure
+// Run runs the installation process, this is done as in a background thread.
-// the required environment, and then spin up the installation container.
+// This will configure the required environment, and then spin up the
-//
+// installation container. Once the container finishes installing the results
-// Once the container finishes installing the results will be stored in an installation
+// are stored in an installation log in the server's configuration directory.
-// log in the server's configuration directory.
 func (ip *InstallationProcess) Run() error {
 	ip.Server.Log().Debug("acquiring installation process lock")
 	if !ip.Server.installing.SwapIf(true) {
@@ -267,9 +264,9 @@ func (ip *InstallationProcess) pullInstallationImage() error {
 		imagePullOptions.RegistryAuth = b64
 	}
 
-	r, err := ip.client.ImagePull(context.Background(), ip.Script.ContainerImage, imagePullOptions)
+	r, err := ip.client.ImagePull(ip.Server.Context(), ip.Script.ContainerImage, imagePullOptions)
 	if err != nil {
-		images, ierr := ip.client.ImageList(context.Background(), types.ImageListOptions{})
+		images, ierr := ip.client.ImageList(ip.Server.Context(), types.ImageListOptions{})
 		if ierr != nil {
 			// Well damn, something has gone really wrong here, just go ahead and abort there
 			// isn't much anything we can do to try and self-recover from this.
@@ -312,9 +309,10 @@ func (ip *InstallationProcess) pullInstallationImage() error {
 	return nil
 }
 
-// Runs before the container is executed. This pulls down the required docker container image
+// BeforeExecute runs before the container is executed. This pulls down the
-// as well as writes the installation script to the disk. This process is executed in an async
+// required docker container image as well as writes the installation script to
-// manner, if either one fails the error is returned.
+// the disk. This process is executed in an async manner, if either one fails
+// the error is returned.
 func (ip *InstallationProcess) BeforeExecute() error {
 	if err := ip.writeScriptToDisk(); err != nil {
 		return errors.WithMessage(err, "failed to write installation script to disk")
@@ -340,7 +338,7 @@ func (ip *InstallationProcess) AfterExecute(containerId string) error {
 	defer ip.RemoveContainer()
 
 	ip.Server.Log().WithField("container_id", containerId).Debug("pulling installation logs for server")
-	reader, err := ip.client.ContainerLogs(ip.context, containerId, types.ContainerLogsOptions{
+	reader, err := ip.client.ContainerLogs(ip.Server.Context(), containerId, types.ContainerLogsOptions{
 		ShowStdout: true,
 		ShowStderr: true,
 		Follow:     false,
@@ -395,12 +393,13 @@ func (ip *InstallationProcess) AfterExecute(containerId string) error {
 	return nil
 }
 
-// Executes the installation process inside a specially created docker container.
+// Execute executes the installation process inside a specially created docker
+// container.
 func (ip *InstallationProcess) Execute() (string, error) {
 	// Create a child context that is canceled once this function is done running. This
 	// will also be canceled if the parent context (from the Server struct) is canceled
 	// which occurs if the server is deleted.
-	ctx, cancel := context.WithCancel(ip.context)
+	ctx, cancel := context.WithCancel(ip.Server.Context())
 	defer cancel()
 
 	conf := &container.Config{
@@ -511,18 +510,15 @@ func (ip *InstallationProcess) Execute() (string, error) {
 // the server configuration directory, as well as to a websocket listener so
 // that the process can be viewed in the panel by administrators.
 func (ip *InstallationProcess) StreamOutput(ctx context.Context, id string) error {
-	reader, err := ip.client.ContainerLogs(ctx, id, types.ContainerLogsOptions{
+	opts := types.ContainerLogsOptions{ShowStdout: true, ShowStderr: true, Follow: true}
-		ShowStdout: true,
+	reader, err := ip.client.ContainerLogs(ctx, id, opts)
-		ShowStderr: true,
-		Follow:     true,
-	})
 	if err != nil {
 		return err
 	}
 	defer reader.Close()
 
-	err = system.ScanReader(reader, ip.Server.Sink(InstallSink).Push)
+	err = system.ScanReader(reader, ip.Server.Sink(system.InstallSink).Push)
-	if err != nil {
+	if err != nil && !errors.Is(err, context.Canceled) {
 		ip.Server.Log().WithFields(log.Fields{"container_id": id, "error": err}).Warn("error processing install output lines")
 	}
 	return nil

server/listeners.go

@@ -5,11 +5,13 @@ import (
 	"regexp"
 	"strconv"
 	"sync"
+	"time"
 
 	"github.com/apex/log"
+	"github.com/pterodactyl/wings/events"
+	"github.com/pterodactyl/wings/system"
 
 	"github.com/pterodactyl/wings/environment"
-	"github.com/pterodactyl/wings/events"
 	"github.com/pterodactyl/wings/remote"
 )
 
@@ -44,7 +46,7 @@ func (dsl *diskSpaceLimiter) Reset() {
 func (dsl *diskSpaceLimiter) Trigger() {
 	dsl.o.Do(func() {
 		dsl.server.PublishConsoleOutputFromDaemon("Server is exceeding the assigned disk space limit, stopping process now.")
-		if err := dsl.server.Environment.WaitForStop(60, true); err != nil {
+		if err := dsl.server.Environment.WaitForStop(dsl.server.Context(), time.Minute, true); err != nil {
 			dsl.server.Log().WithField("error", err).Error("failed to stop server after exceeding space limit!")
 		}
 	})
@@ -72,47 +74,57 @@ func (s *Server) processConsoleOutputEvent(v []byte) {
 		return
 	}
 
-	s.Sink(LogSink).Push(v)
+	s.Sink(system.LogSink).Push(v)
 }
 
 // StartEventListeners adds all the internal event listeners we want to use for
 // a server. These listeners can only be removed by deleting the server as they
 // should last for the duration of the process' lifetime.
 func (s *Server) StartEventListeners() {
-	state := make(chan events.Event)
+	c := make(chan []byte, 8)
-	stats := make(chan events.Event)
+	limit := newDiskLimiter(s)
-	docker := make(chan events.Event)
+
+	s.Log().Debug("registering event listeners: console, state, resources...")
+	s.Environment.Events().On(c)
+	s.Environment.SetLogCallback(s.processConsoleOutputEvent)
 
 	go func() {
-		l := newDiskLimiter(s)
-
 		for {
 			select {
-			case e := <-state:
+			case v := <-c:
-				go func() {
+				go func(v []byte, limit *diskSpaceLimiter) {
-					// Reset the throttler when the process is started.
+					var e events.Event
-					if e.Data == environment.ProcessStartingState {
+					if err := events.DecodeTo(v, &e); err != nil {
-						l.Reset()
+						return
-						s.Throttler().Reset()
 					}
-
+					switch e.Topic {
-					s.OnStateChange()
+					case environment.ResourceEvent:
-				}()
+						{
-			case e := <-stats:
+							var stats struct {
-				go func() {
+								Topic string
-					s.resources.UpdateStats(e.Data.(environment.Stats))
+								Data  environment.Stats
-
+							}
+							if err := events.DecodeTo(v, &stats); err != nil {
+								s.Log().WithField("error", err).Warn("failed to decode server resource event")
+								return
+							}
+							s.resources.UpdateStats(stats.Data)
 							// If there is no disk space available at this point, trigger the server
 							// disk limiter logic which will start to stop the running instance.
 							if !s.Filesystem().HasSpaceAvailable(true) {
-						l.Trigger()
+								limit.Trigger()
 							}
-
 							s.Events().Publish(StatsEvent, s.Proc())
-				}()
+						}
-			case e := <-docker:
+					case environment.StateChangeEvent:
-				go func() {
+						{
-					switch e.Topic {
+							// Reset the throttler when the process is started.
+							if e.Data == environment.ProcessStartingState {
+								limit.Reset()
+								s.Throttler().Reset()
+							}
+							s.OnStateChange()
+						}
 					case environment.DockerImagePullStatus:
 						s.Events().Publish(InstallOutputEvent, e.Data)
 					case environment.DockerImagePullStarted:
@@ -120,18 +132,13 @@ func (s *Server) StartEventListeners() {
 					case environment.DockerImagePullCompleted:
 						s.PublishConsoleOutputFromDaemon("Finished pulling Docker container image")
 					default:
-						s.Log().WithField("topic", e.Topic).Error("unhandled docker event topic")
 					}
-				}()
+				}(v, limit)
+			case <-s.Context().Done():
+				return
 			}
 		}
 	}()
-
-	s.Log().Debug("registering event listeners: console, state, resources...")
-	s.Environment.SetLogCallback(s.processConsoleOutputEvent)
-	s.Environment.Events().On(state, environment.StateChangeEvent)
-	s.Environment.Events().On(stats, environment.ResourceEvent)
-	s.Environment.Events().On(docker, dockerEvents...)
 }
 
 var stripAnsiRegex = regexp.MustCompile("[\u001B\u009B][[\\]()#;?]*(?:(?:(?:[a-zA-Z\\d]*(?:;[a-zA-Z\\d]*)*)?\u0007)|(?:(?:\\d{1,4}(?:;\\d{0,4})*)?[\\dA-PRZcf-ntqry=><~]))")

server/power.go

@@ -133,11 +133,11 @@ func (s *Server) HandlePowerAction(action PowerAction, waitSeconds ...int) error
 
 		return s.Environment.Start(s.Context())
 	case PowerActionStop:
-		// We're specifically waiting for the process to be stopped here, otherwise the lock is released
+		fallthrough
-		// too soon, and you can rack up all sorts of issues.
-		return s.Environment.WaitForStop(10*60, true)
 	case PowerActionRestart:
-		if err := s.Environment.WaitForStop(10*60, true); err != nil {
+		// We're specifically waiting for the process to be stopped here, otherwise the lock is
+		// released too soon, and you can rack up all sorts of issues.
+		if err := s.Environment.WaitForStop(s.Context(), time.Minute*10, true); err != nil {
 			// Even timeout errors should be bubbled back up the stack. If the process didn't stop
 			// nicely, but the terminate argument was passed then the server is stopped without an
 			// error being returned.
@@ -149,6 +149,10 @@ func (s *Server) HandlePowerAction(action PowerAction, waitSeconds ...int) error
 			return err
 		}
 
+		if action == PowerActionStop {
+			return nil
+		}
+
 		// Now actually try to start the process by executing the normal pre-boot logic.
 		if err := s.onBeforeStart(); err != nil {
 			return err
@@ -156,7 +160,7 @@ func (s *Server) HandlePowerAction(action PowerAction, waitSeconds ...int) error
 
 		return s.Environment.Start(s.Context())
 	case PowerActionTerminate:
-		return s.Environment.Terminate(os.Kill)
+		return s.Environment.Terminate(s.Context(), os.Kill)
 	}
 
 	return errors.New("attempting to handle unknown power action")
@@ -197,15 +201,19 @@ func (s *Server) onBeforeStart() error {
 	// we don't need to actively do anything about it at this point, worse comes to worst the
 	// server starts in a weird state and the user can manually adjust.
 	s.PublishConsoleOutputFromDaemon("Updating process configuration files...")
+	s.Log().Debug("updating server configuration files...")
 	s.UpdateConfigurationFiles()
+	s.Log().Debug("updated server configuration files")
 
 	if config.Get().System.CheckPermissionsOnBoot {
 		s.PublishConsoleOutputFromDaemon("Ensuring file permissions are set correctly, this could take a few seconds...")
 		// Ensure all the server file permissions are set correctly before booting the process.
+		s.Log().Debug("chowning server root directory...")
 		if err := s.Filesystem().Chown("/"); err != nil {
 			return errors.WithMessage(err, "failed to chown root server directory during pre-boot process")
 		}
 	}
 
+	s.Log().Info("completed server preflight, starting boot process...")
 	return nil
 }

server/server.go

@@ -70,10 +70,10 @@ type Server struct {
 	wsBag       *WebsocketBag
 	wsBagLocker sync.Mutex
 
-	sinks map[SinkName]*sinkPool
+	sinks map[system.SinkName]*system.SinkPool
 
-	logSink     *sinkPool
+	logSink     *system.SinkPool
-	installSink *sinkPool
+	installSink *system.SinkPool
 }
 
 // New returns a new server instance with a context and all of the default
@@ -88,9 +88,9 @@ func New(client remote.Client) (*Server, error) {
 		transferring: system.NewAtomicBool(false),
 		restoring:    system.NewAtomicBool(false),
 		powerLock:    system.NewLocker(),
-		sinks: map[SinkName]*sinkPool{
+		sinks: map[system.SinkName]*system.SinkPool{
-			LogSink:     newSinkPool(),
+			system.LogSink:     system.NewSinkPool(),
-			InstallSink: newSinkPool(),
+			system.InstallSink: system.NewSinkPool(),
 		},
 	}
 	if err := defaults.Set(&s); err != nil {

server/update.go

@@ -1,6 +1,8 @@
 package server
 
 import (
+	"time"
+
 	"github.com/pterodactyl/wings/environment/docker"
 
 	"github.com/pterodactyl/wings/environment"
@@ -58,7 +60,7 @@ func (s *Server) SyncWithEnvironment() {
 			s.Log().Info("server suspended with running process state, terminating now")
 
 			go func(s *Server) {
-				if err := s.Environment.WaitForStop(60, true); err != nil {
+				if err := s.Environment.WaitForStop(s.Context(), time.Minute, true); err != nil {
 					s.Log().WithField("error", err).Warn("failed to terminate server environment after suspension")
 				}
 			}(s)

system/locker.go

@@ -51,7 +51,9 @@ func (l *Locker) TryAcquire(ctx context.Context) error {
 		return nil
 	case <-ctx.Done():
 		if err := ctx.Err(); err != nil {
-			return err
+			if errors.Is(err, context.DeadlineExceeded) || errors.Is(err, context.Canceled) {
+				return ErrLockerLocked
+			}
 		}
 		return nil
 	}

system/locker_test.go

@@ -81,7 +81,7 @@ func TestPower(t *testing.T) {
 				err := l.TryAcquire(ctx)
 
 				g.Assert(err).IsNotNil()
-				g.Assert(errors.Is(err, context.DeadlineExceeded)).IsTrue()
+				g.Assert(errors.Is(err, ErrLockerLocked)).IsTrue()
 				g.Assert(cap(l.ch)).Equal(1)
 				g.Assert(len(l.ch)).Equal(1)
 				g.Assert(l.IsLocked()).IsTrue()

system/sink_pool.go

@@ -1,4 +1,4 @@
-package server
+package system
 
 import (
 	"sync"
@@ -16,20 +16,20 @@ const (
 	InstallSink SinkName = "install"
 )
 
-// sinkPool represents a pool with sinks.
+// SinkPool represents a pool with sinks.
-type sinkPool struct {
+type SinkPool struct {
 	mu    sync.RWMutex
 	sinks []chan []byte
 }
 
-// newSinkPool returns a new empty sinkPool. A sink pool generally lives with a
+// NewSinkPool returns a new empty SinkPool. A sink pool generally lives with a
 // server instance for it's full lifetime.
-func newSinkPool() *sinkPool {
+func NewSinkPool() *SinkPool {
-	return &sinkPool{}
+	return &SinkPool{}
 }
 
 // On adds a channel to the sink pool instance.
-func (p *sinkPool) On(c chan []byte) {
+func (p *SinkPool) On(c chan []byte) {
 	p.mu.Lock()
 	p.sinks = append(p.sinks, c)
 	p.mu.Unlock()
@@ -37,7 +37,7 @@ func (p *sinkPool) On(c chan []byte) {
 
 // Off removes a given channel from the sink pool. If no matching sink is found
 // this function is a no-op. If a matching channel is found, it will be removed.
-func (p *sinkPool) Off(c chan []byte) {
+func (p *SinkPool) Off(c chan []byte) {
 	p.mu.Lock()
 	defer p.mu.Unlock()
 
@@ -66,7 +66,7 @@ func (p *sinkPool) Off(c chan []byte) {
 
 // Destroy destroys the pool by removing and closing all sinks and destroying
 // all of the channels that are present.
-func (p *sinkPool) Destroy() {
+func (p *SinkPool) Destroy() {
 	p.mu.Lock()
 	defer p.mu.Unlock()
 
@@ -95,7 +95,7 @@ func (p *sinkPool) Destroy() {
 // likely the best option anyways. This uses waitgroups to allow every channel
 // to attempt its send concurrently thus making the total blocking time of this
 // function "O(1)" instead of "O(n)".
-func (p *sinkPool) Push(data []byte) {
+func (p *SinkPool) Push(data []byte) {
 	p.mu.RLock()
 	defer p.mu.RUnlock()
 	var wg sync.WaitGroup
@@ -119,24 +119,3 @@ func (p *sinkPool) Push(data []byte) {
 	}
 	wg.Wait()
 }
-
-// Sink returns the instantiated and named sink for a server. If the sink has
-// not been configured yet this function will cause a panic condition.
-func (s *Server) Sink(name SinkName) *sinkPool {
-	sink, ok := s.sinks[name]
-	if !ok {
-		s.Log().Fatalf("attempt to access nil sink: %s", name)
-	}
-	return sink
-}
-
-// DestroyAllSinks iterates over all of the sinks configured for the server and
-// destroys their instances. Note that this will cause a panic if you attempt
-// to call Server.Sink() again after. This function is only used when a server
-// is being deleted from the system.
-func (s *Server) DestroyAllSinks() {
-	s.Log().Info("destroying all registered sinks for server instance")
-	for _, sink := range s.sinks {
-		sink.Destroy()
-	}
-}

system/sink_pool_test.go

@@ -1,4 +1,4 @@
-package server
+package system
 
 import (
 	"fmt"
@@ -23,7 +23,7 @@ func TestSink(t *testing.T) {
 
 	g.Describe("SinkPool#On", func() {
 		g.It("pushes additional channels to a sink", func() {
-			pool := &sinkPool{}
+			pool := &SinkPool{}
 
 			g.Assert(pool.sinks).IsZero()
 
@@ -36,9 +36,9 @@ func TestSink(t *testing.T) {
 	})
 
 	g.Describe("SinkPool#Off", func() {
-		var pool *sinkPool
+		var pool *SinkPool
 		g.BeforeEach(func() {
-			pool = &sinkPool{}
+			pool = &SinkPool{}
 		})
 
 		g.It("works when no sinks are registered", func() {
@@ -97,9 +97,9 @@ func TestSink(t *testing.T) {
 	})
 
 	g.Describe("SinkPool#Push", func() {
-		var pool *sinkPool
+		var pool *SinkPool
 		g.BeforeEach(func() {
-			pool = &sinkPool{}
+			pool = &SinkPool{}
 		})
 
 		g.It("works when no sinks are registered", func() {
@@ -190,9 +190,9 @@ func TestSink(t *testing.T) {
 	})
 
 	g.Describe("SinkPool#Destroy", func() {
-		var pool *sinkPool
+		var pool *SinkPool
 		g.BeforeEach(func() {
-			pool = &sinkPool{}
+			pool = &SinkPool{}
 		})
 
 		g.It("works if no sinks are registered", func() {

system/utils.go

@@ -3,12 +3,10 @@ package system
 import (
 	"bufio"
 	"bytes"
-	"context"
 	"fmt"
 	"io"
 	"strconv"
 	"sync"
-	"time"
 
 	"emperror.dev/errors"
 	"github.com/goccy/go-json"
@@ -90,16 +88,16 @@ func ScanReader(r io.Reader, callback func(line []byte)) error {
 			} else {
 				buf.Write(line)
 			}
+			// If we encountered an error with something in ReadLine that was not an
+			// EOF just abort the entire process here.
+			if err != nil && err != io.EOF {
+				return err
+			}
 			// Finish this loop and begin outputting the line if there is no prefix
 			// (the line fit into the default buffer), or if we hit the end of the line.
 			if !isPrefix || err == io.EOF {
 				break
 			}
-			// If we encountered an error with something in ReadLine that was not an
-			// EOF just abort the entire process here.
-			if err != nil {
-				return err
-			}
 		}
 
 		// Send the full buffer length over to the event handler to be emitted in
@@ -122,22 +120,6 @@ func ScanReader(r io.Reader, callback func(line []byte)) error {
 	return nil
 }
 
-// Runs a given work function every "d" duration until the provided context is canceled.
-func Every(ctx context.Context, d time.Duration, work func(t time.Time)) {
-	ticker := time.NewTicker(d)
-	go func() {
-		for {
-			select {
-			case <-ctx.Done():
-				ticker.Stop()
-				return
-			case t := <-ticker.C:
-				work(t)
-			}
-		}
-	}()
-}
-
 func FormatBytes(b int64) string {
 	if b < 1024 {
 		return fmt.Sprintf("%d B", b)