Collect all error messages before stopping pgstream

pkg/wal/processor/batch/wal_batch_sender.go

@@ -22,7 +22,7 @@ type Sender[T Message] struct {
 	queueBytesSema   synclib.WeightedSemaphore
 	msgChan          chan (*WALMessage[T])
 	once             *sync.Once
-	sendDone         chan (error)
+	sendDone         chan struct{}
 	sendErr          error
 	ignoreSendErrors bool
 
@@ -47,7 +47,7 @@ func NewSender[T Message](ctx context.Context, config *Config, sendfn sendBatchF
 		maxBatchBytes:     config.GetMaxBatchBytes(),
 		maxBatchSize:      config.GetMaxBatchSize(),
 		msgChan:           make(chan *WALMessage[T]),
-		sendDone:          make(chan error, 1),
+		sendDone:          make(chan struct{}),
 		once:              &sync.Once{},
 		logger:            logger,
 		sendBatchFn:       sendfn,
@@ -106,12 +106,11 @@ func (s *Sender[T]) SendMessage(ctx context.Context, msg *WALMessage[T]) error {
 	// longer processing), and an error will be returned.
 	select {
 	case s.msgChan <- msg:
-	case sendDoneErr, ok := <-s.sendDone:
-		// check if a different call has closed the send channel already, to
-		// prevent blocking when called concurrently.
-		if ok && sendDoneErr != nil {
-			s.sendErr = sendDoneErr
-		}
+	case <-s.sendDone:
+		// s.sendErr is set by send() before closing s.sendDone, so it is safe
+		// to read here from any number of concurrent callers. It is always
+		// non-nil — every return path of batchMsgLoop carries an error
+		// (ctx.Err, sendErrChan, or a non-nil drainBatch result).
 		s.logger.Error(s.sendErr, "stop processing, sending has stopped")
 		return fmt.Errorf("%w: %w", errSendStopped, s.sendErr)
 	}
@@ -211,7 +210,9 @@ func (s *Sender[T]) send(ctx context.Context) error {
 	if err != nil && !errors.Is(err, context.Canceled) {
 		s.logger.Error(err, "sending stopped")
 	}
-	s.sendDone <- err
+	// publish the send error before signalling shutdown so any goroutines
+	// waiting in SendMessage can observe it after the channel is closed.
+	s.sendErr = err
 	close(s.sendDone)
 	return err
 }

pkg/wal/processor/batch/wal_batch_sender_test.go

@@ -124,7 +124,7 @@ func TestSender_SendMessage(t *testing.T) {
 				maxBatchSize:      10,
 				msgChan:           make(chan *WALMessage[*mockMessage]),
 				queueBytesSema:    tc.weightedSemaphore,
-				sendDone:          make(chan error, 1),
+				sendDone:          make(chan struct{}),
 				once:              &sync.Once{},
 				logger:            log.NewNoopLogger(),
 				sendBatchFn:       noopSendFn,
@@ -137,7 +137,9 @@ func TestSender_SendMessage(t *testing.T) {
 			defer cancel()
 
 			if tc.sendDone {
-				batchSender.sendDone <- errSendStopped
+				// Simulate the post-fix shape: send() publishes the error
+				// into the shared field before closing sendDone.
+				batchSender.sendErr = errSendStopped
 				close(batchSender.sendDone)
 			}
 
@@ -309,7 +311,7 @@ func TestSender_send(t *testing.T) {
 				maxBatchSize:      10,
 				msgChan:           make(chan *WALMessage[*mockMessage]),
 				queueBytesSema:    tc.semaphore,
-				sendDone:          make(chan error, 1),
+				sendDone:          make(chan struct{}),
 				once:              &sync.Once{},
 				logger:            log.NewNoopLogger(),
 				sendBatchFn:       tc.sendFn(doneChan),
@@ -382,7 +384,7 @@ func TestSender_send(t *testing.T) {
 					}
 				},
 			},
-			sendDone:    make(chan error, 1),
+			sendDone:    make(chan struct{}),
 			once:        &sync.Once{},
 			logger:      log.NewNoopLogger(),
 			sendBatchFn: sendFn(doneChan),
@@ -464,3 +466,76 @@ func TestSender(t *testing.T) {
 		}
 	}
 }
+
+// Regression test for https://github.com/xataio/pgstream/issues/372:
+// concurrent SendMessage callers must all observe the underlying send error
+// rather than wrapping a nil sendErr (which produced "%!w(<nil>)" messages
+// that obscured the real cause of snapshot worker failures).
+func TestSender_ConcurrentSendErrorPropagation(t *testing.T) {
+	t.Parallel()
+
+	ctx := context.Background()
+
+	errTest := errors.New("oh noes")
+	testCommitPos := wal.CommitPosition("1")
+
+	mockMsg := func(i uint) *mockMessage {
+		return &mockMessage{id: i}
+	}
+	testWALMsg := func(i uint) *WALMessage[*mockMessage] {
+		return NewWALMessage(mockMsg(i), testCommitPos)
+	}
+
+	doneChan := make(chan struct{}, 1)
+	defer close(doneChan)
+
+	sendFn := func(doneChan chan<- struct{}) sendBatchFn[*mockMessage] {
+		once := sync.Once{}
+		return func(ctx context.Context, b *Batch[*mockMessage]) error {
+			defer once.Do(func() { doneChan <- struct{}{} })
+			return errTest
+		}
+	}
+
+	sender, err := NewSender(ctx, &Config{
+		BatchTimeout: 100 * time.Millisecond,
+		MaxBatchSize: 1,
+	}, sendFn(doneChan), log.NewNoopLogger())
+	require.NoError(t, err)
+	defer sender.Close()
+
+	// prime the sender so the batch send fails
+	require.NoError(t, sender.SendMessage(ctx, testWALMsg(1)))
+
+	select {
+	case <-doneChan:
+	case <-time.After(5 * time.Second):
+		t.Fatal("timed out waiting for send to fail")
+	}
+	// Wait deterministically for send() to publish the error and close
+	// sendDone — that's the exact "happens-before" we want every concurrent
+	// SendMessage caller below to observe.
+	select {
+	case <-sender.sendDone:
+	case <-time.After(5 * time.Second):
+		t.Fatal("timed out waiting for sendDone to close")
+	}
+
+	const workers = 8
+	errs := make([]error, workers)
+	wg := sync.WaitGroup{}
+	wg.Add(workers)
+	for i := 0; i < workers; i++ {
+		go func(i int) {
+			defer wg.Done()
+			errs[i] = sender.SendMessage(ctx, testWALMsg(uint(i+2)))
+		}(i)
+	}
+	wg.Wait()
+
+	for i, err := range errs {
+		require.ErrorIsf(t, err, errSendStopped, "worker %d: missing errSendStopped", i)
+		require.ErrorIsf(t, err, errTest, "worker %d: missing underlying send error", i)
+		require.NotContainsf(t, err.Error(), "%!w(<nil>)", "worker %d: nil error wrapping leaked through", i)
+	}
+}

pkg/wal/processor/webhook/notifier/webhook_notifier.go

@@ -36,9 +36,13 @@ type Notifier struct {
 	queueBytesSema synclib.WeightedSemaphore
 	notifyChan     chan *notifyMsg
 	workerCount    uint
-	notifyDone     chan (error)
-	notifyErr      error
-	once           *sync.Once
+	// shutdownCh is closed by Close() to signal Notify and any in-flight
+	// ProcessWALEvent calls to stop. notifyChan is never closed, so concurrent
+	// senders cannot panic on "send on closed channel".
+	shutdownCh chan struct{}
+	notifyDone chan struct{}
+	notifyErr  error
+	once       *sync.Once
 }
 
 type subscriptionRetriever interface {
@@ -59,7 +63,8 @@ func New(cfg *Config, store subscriptionRetriever, opts ...Option) *Notifier {
 		notifyChan:        make(chan *notifyMsg),
 		workerCount:       cfg.workerCount(),
 		serialiser:        json.Marshal,
-		notifyDone:        make(chan error, 1),
+		shutdownCh:        make(chan struct{}),
+		notifyDone:        make(chan struct{}),
 		once:              &sync.Once{},
 	}
 
@@ -130,11 +135,19 @@ func (n *Notifier) ProcessWALEvent(ctx context.Context, walEvent *wal.Event) (er
 
 	select {
 	case n.notifyChan <- msg:
-	case notifyDoneErr, ok := <-n.notifyDone:
-		if ok && notifyDoneErr != nil {
-			n.notifyErr = notifyDoneErr
-		}
+	case <-n.shutdownCh:
+		// Close() was called before Notify processed this event. notifyChan is
+		// never closed, so we cannot send into it — bail out cleanly.
+		n.logger.Error(nil, "stop processing, notify is shutting down")
+		return errNotifyStopped
+	case <-n.notifyDone:
+		// Notify has exited on its own (external ctx cancel or notify error).
+		// n.notifyErr is set by Notify before closing n.notifyDone, so it is
+		// safe to read here from any number of concurrent callers.
 		n.logger.Error(n.notifyErr, "stop processing, notify has stopped")
+		if n.notifyErr == nil {
+			return errNotifyStopped
+		}
 		return fmt.Errorf("%w: %w", errNotifyStopped, n.notifyErr)
 	}
 
@@ -147,6 +160,9 @@ func (n *Notifier) Notify(ctx context.Context) error {
 			select {
 			case <-ctx.Done():
 				return ctx.Err()
+			case <-n.shutdownCh:
+				// graceful shutdown via Close(); not an error
+				return nil
 			case msg := <-n.notifyChan:
 				err := n.notify(ctx, msg)
 				n.queueBytesSema.Release(int64(msg.size()))
@@ -163,7 +179,9 @@ func (n *Notifier) Notify(ctx context.Context) error {
 	}
 
 	err := notifyLoop()
-	n.notifyDone <- err
+	// publish the notify error before signalling shutdown so any goroutines
+	// waiting in ProcessWALEvent can observe it after the channel is closed.
+	n.notifyErr = err
 	close(n.notifyDone)
 	return err
 }
@@ -172,17 +190,14 @@ func (n *Notifier) Name() string {
 	return "webhooks-notifier"
 }
 
+// Close signals Notify and any in-flight ProcessWALEvent callers to stop. It
+// is safe to call multiple times. notifyChan itself is not closed: that would
+// race with a concurrent ProcessWALEvent's send and panic.
 func (n *Notifier) Close() error {
-	n.closeNotifyChan()
-	return nil
-}
-
-// closeNotifyChan closes the internal notify channel. It can be called multiple
-// times.
-func (n *Notifier) closeNotifyChan() {
 	n.once.Do(func() {
-		close(n.notifyChan)
+		close(n.shutdownCh)
 	})
+	return nil
 }
 
 func (n *Notifier) notify(ctx context.Context, msg *notifyMsg) error {