Fix goroutine leaks, panic recovery, and zombie process

Author: dcadenas

cmd/event-service/di/service.go

@@ -2,9 +2,12 @@ package di
 
 import (
 	"context"
+	"os"
+	"time"
 
 	"github.com/boreq/errors"
 	"github.com/hashicorp/go-multierror"
+	"github.com/planetary-social/nos-event-service/internal/goroutine"
 	"github.com/planetary-social/nos-event-service/internal/logging"
 	"github.com/planetary-social/nos-event-service/internal/migrations"
 	"github.com/planetary-social/nos-event-service/service/adapters/sqlite"
@@ -72,6 +75,8 @@ func (s Service) ExecuteMigrations(ctx context.Context) error {
 	return s.migrationsRunner.Run(ctx, s.migrations, s.migrationsProgressCallback)
 }
 
+const shutdownTimeout = 30 * time.Second
+
 func (s Service) Run(ctx context.Context) error {
 	ctx, cancel := context.WithCancel(ctx)
 	defer cancel()
@@ -81,66 +86,83 @@ func (s Service) Run(ctx context.Context) error {
 
 	runners++
 	// Serve http
-	go func() {
-		errCh <- errors.Wrap(s.server.ListenAndServe(ctx), "server error")
-	}()
+	goroutine.Run(s.logger, errCh, func() error {
+		return errors.Wrap(s.server.ListenAndServe(ctx), "server error")
+	})
 
 	// Fetch events from the database relays and send them to the in memory pubsub
 	runners++
-	go func() {
-		errCh <- errors.Wrap(s.downloader.Run(ctx), "downloader error")
-	}()
+	goroutine.Run(s.logger, errCh, func() error {
+		return errors.Wrap(s.downloader.Run(ctx), "downloader error")
+	})
 
 	// Subscribe to the in memory pubsub of events, emit a NewSaveReceivedEvent
 	// command that will make some checks on the event, save it if the check
 	// passes and emit a EventSavedEvent to the sqlite pubsub.
 	runners++
-	go func() {
-		errCh <- errors.Wrap(s.receivedEventSubscriber.Run(ctx), "received event subscriber error")
-	}()
+	goroutine.Run(s.logger, errCh, func() error {
+		return errors.Wrap(s.receivedEventSubscriber.Run(ctx), "received event subscriber error")
+	})
 
 	// Subscribe to saved events in the database. This uses the sqlite pubsub. This triggers:
 	// - analysis to extract new relays and store them in db. They will be used by the downloader.
 	// - analysis to store pubkeys and store them in the db (contacts_followees, pubkeys, contacts_events). This will be used by the downloader.
 	// - publish to watermill pubsub
 	// - may publish the event in wss://relay.nos.social if they are metadata related
 	runners++
-	go func() {
-		errCh <- errors.Wrap(s.eventSavedEventSubscriber.Run(ctx), "event saved subscriber error")
-	}()
+	goroutine.Run(s.logger, errCh, func() error {
+		return errors.Wrap(s.eventSavedEventSubscriber.Run(ctx), "event saved subscriber error")
+	})
 
 	// The metrics timer collects metrics from the app.
 	runners++
-	go func() {
-		errCh <- errors.Wrap(s.metricsTimer.Run(ctx), "metrics timer error")
-	}()
+	goroutine.Run(s.logger, errCh, func() error {
+		return errors.Wrap(s.metricsTimer.Run(ctx), "metrics timer error")
+	})
 
 	// Sqlite transaction runner
 	runners++
-	go func() {
-		errCh <- errors.Wrap(s.transactionRunner.Run(ctx), "transaction runner error")
-	}()
+	goroutine.Run(s.logger, errCh, func() error {
+		return errors.Wrap(s.transactionRunner.Run(ctx), "transaction runner error")
+	})
 
 	// The task scheduler creates sequential time window based tasks that
 	// contain filters to be applied to each relay to fetch the events we want.
 	// Event downloaders subscribe to this.
 	runners++
-	go func() {
-		errCh <- errors.Wrap(s.taskScheduler.Run(ctx), "task scheduler error")
-	}()
+	goroutine.Run(s.logger, errCh, func() error {
+		return errors.Wrap(s.taskScheduler.Run(ctx), "task scheduler error")
+	})
 
 	// The cleanup timer periodically removes processed events older than retention period
 	runners++
-	go func() {
-		errCh <- errors.Wrap(s.cleanupTimer.Run(ctx), "cleanup timer error")
-	}()
+	goroutine.Run(s.logger, errCh, func() error {
+		return errors.Wrap(s.cleanupTimer.Run(ctx), "cleanup timer error")
+	})
+
+	// Wait for the first runner to exit.
+	firstErr := <-errCh
+	s.logger.Error().WithError(firstErr).Message("first runner terminated, shutting down")
+	cancel()
 
+	// Collect remaining errors with a timeout to prevent zombie processes.
 	var compoundErr error
-	for i := 0; i < runners; i++ {
-		err := errors.Wrap(<-errCh, "error returned by runner")
-		s.logger.Error().WithError(err).Message("runner terminated")
-		compoundErr = multierror.Append(compoundErr, err)
-		cancel()
+	compoundErr = multierror.Append(compoundErr, errors.Wrap(firstErr, "error returned by runner"))
+
+	timer := time.NewTimer(shutdownTimeout)
+	defer timer.Stop()
+
+	for i := 1; i < runners; i++ {
+		select {
+		case err := <-errCh:
+			err = errors.Wrap(err, "error returned by runner")
+			s.logger.Error().WithError(err).Message("runner terminated")
+			compoundErr = multierror.Append(compoundErr, err)
+		case <-timer.C:
+			s.logger.Error().Message("shutdown timeout exceeded, forcing exit")
+			os.Exit(1)
+		}
 	}
+
 	return compoundErr
 }

internal/goroutine/goroutine.go

@@ -0,0 +1,27 @@
+package goroutine
+
+import (
+	"fmt"
+	"runtime/debug"
+
+	"github.com/planetary-social/nos-event-service/internal/logging"
+)
+
+// Run starts a goroutine that executes fn and sends its return value to errCh.
+// If fn panics, the panic is recovered, logged with a full stack trace, and
+// converted to an error sent to errCh. This prevents goroutines from silently
+// dying and leaving the service in a zombie state.
+func Run(logger logging.Logger, errCh chan<- error, fn func() error) {
+	go func() {
+		defer func() {
+			if r := recover(); r != nil {
+				stack := string(debug.Stack())
+				err := fmt.Errorf("goroutine panicked: %v\n%s", r, stack)
+				logger.Error().WithError(err).Message("recovered from panic in goroutine")
+				errCh <- err
+			}
+		}()
+
+		errCh <- fn()
+	}()
+}

internal/goroutine/goroutine_test.go

@@ -0,0 +1,60 @@
+package goroutine_test
+
+import (
+	"strings"
+	"testing"
+
+	"github.com/boreq/errors"
+	"github.com/planetary-social/nos-event-service/internal/goroutine"
+	"github.com/planetary-social/nos-event-service/internal/logging"
+	"github.com/stretchr/testify/assert"
+	"github.com/stretchr/testify/require"
+)
+
+func TestRun_PropagatesError(t *testing.T) {
+	errCh := make(chan error, 1)
+	expectedErr := errors.New("test error")
+
+	goroutine.Run(logging.NewDevNullLogger(), errCh, func() error {
+		return expectedErr
+	})
+
+	err := <-errCh
+	assert.ErrorIs(t, err, expectedErr)
+}
+
+func TestRun_RecoversPanic(t *testing.T) {
+	errCh := make(chan error, 1)
+
+	goroutine.Run(logging.NewDevNullLogger(), errCh, func() error {
+		panic("something went wrong")
+	})
+
+	err := <-errCh
+	require.Error(t, err)
+	assert.Contains(t, err.Error(), "goroutine panicked: something went wrong")
+	assert.Contains(t, err.Error(), "goroutine.go")
+}
+
+func TestRun_RecoversPanicWithError(t *testing.T) {
+	errCh := make(chan error, 1)
+
+	goroutine.Run(logging.NewDevNullLogger(), errCh, func() error {
+		panic(errors.New("panic error"))
+	})
+
+	err := <-errCh
+	require.Error(t, err)
+	assert.True(t, strings.Contains(err.Error(), "panic error"))
+}
+
+func TestRun_SuccessfulCompletion(t *testing.T) {
+	errCh := make(chan error, 1)
+
+	goroutine.Run(logging.NewDevNullLogger(), errCh, func() error {
+		return nil
+	})
+
+	err := <-errCh
+	assert.NoError(t, err)
+}

service/domain/relays/relay_connection.go

@@ -150,7 +150,7 @@ func (r *RelayConnection) GetEvents(ctx context.Context, filter domain.Filter) (
 		// Context is cancelled from a task.CheckIfDoneAndEnd() triggered from EOSE messages
 		<-ctx.Done()
 		if err := r.removeSubscriptionChannel(ch); err != nil {
-			panic(err)
+			r.logger.Error().WithError(err).Message("error removing subscription channel")
 		}
 	}()
 
@@ -176,7 +176,7 @@ func (r *RelayConnection) SendEvent(ctx context.Context, event domain.Event) err
 
 	defer func() {
 		if err := r.removeEventChannel(event, ch); err != nil {
-			panic(err)
+			r.logger.Error().WithError(err).Message("error removing event channel")
 		}
 	}()
 
@@ -303,12 +303,18 @@ func (r *RelayConnection) run(ctx context.Context) error {
 	r.setState(RelayConnectionStateConnected)
 	r.logger.Trace().Message("connected")
 
+	var wg sync.WaitGroup
+
+	wg.Add(1)
 	go func() {
+		defer wg.Done()
 		<-ctx.Done()
 		_ = conn.Close()
 	}()
 
+	wg.Add(1)
 	go func() {
+		defer wg.Done()
 		if err := r.manageSubs(ctx, conn); err != nil {
 			if !r.writeErrorShouldNotBeLogged(err) {
 				r.logger.Error().
@@ -318,7 +324,9 @@ func (r *RelayConnection) run(ctx context.Context) error {
 		}
 	}()
 
+	wg.Add(1)
 	go func() {
+		defer wg.Done()
 		// Loops through newEventsCh and sends events to the relay
 		if err := r.sendEvents(ctx, conn); err != nil {
 			if !r.writeErrorShouldNotBeLogged(err) {
@@ -329,15 +337,17 @@ func (r *RelayConnection) run(ctx context.Context) error {
 		}
 	}()
 
+	var readErr error
 	for {
 		messageBytes, err := conn.ReadMessage()
 		if err != nil {
-			return NewReadMessageError(err)
+			readErr = NewReadMessageError(err)
+			break
 		}
 
 		if r.state == RelayConnectionStateBackPressured {
-			// Load shedding under backpressure, we just disconnect
-			return BackPressureError
+			readErr = BackPressureError
+			break
 		}
 
 		if err := r.handleMessage(messageBytes); err != nil {
@@ -348,6 +358,10 @@ func (r *RelayConnection) run(ctx context.Context) error {
 				Message("error handling an incoming message")
 		}
 	}
+
+	cancel()
+	wg.Wait()
+	return readErr
 }
 
 func (r *RelayConnection) writeErrorShouldNotBeLogged(err error) bool {