Drop too large WAL events

listener/listener.go

@@ -555,50 +555,7 @@ func (l *Listener) Stream(ctx context.Context) error {
 				l.log.Warn("stream: context canceled", "err", ctx.Err())
 				return nil
 			case object := <-resultChan:
-				subjectName, result, event, xld, pkm := object.subjectName, object.result, object.event, object.xld, object.pkm
-
-				if result != nil {
-					_, err := result.Get(ctx)
-					if err != nil {
-						// Only warn on publish failures, but continue to process and ack messages
-						// this runs the risk of losing data, but this it is more important to keep processing WAL messages in the meantime
-						l.monitor.IncProblematicEvents(problemKindPublish)
-
-						l.log.Warn(
-							"failed to publish message",
-							slog.Any("error", err),
-							slog.String("subjectName", subjectName),
-							slog.String("table", event.Table),
-							slog.String("action", event.Action),
-						)
-					} else {
-						l.monitor.IncPublishedEvents(subjectName, event.Table)
-						l.log.Debug(
-							"event was sent",
-							slog.String("subject", subjectName),
-							slog.String("action", event.Action),
-							slog.String("schema", event.Schema),
-							slog.String("table", event.Table),
-							slog.String("lsn", l.readLSN().String()),
-						)
-					}
-					tx.pool.Put(event)
-				}
-
-				latest := latestWalStart.Load()
-				if pkm != nil {
-					walEnd := uint64(pkm.ServerWALEnd)
-					if walEnd > latest {
-						latestWalStart.Store(walEnd)
-					}
-				} else if xld != nil {
-					// We do not need to compare and swap here as there's only one thread
-					// writing to this value
-					walStart := uint64(xld.WALStart)
-					if xld.WALData != nil && walStart > latest {
-						latestWalStart.Store(walStart)
-					}
-				}
+				l.handlePublishResult(ctx, object, tx.pool, &latestWalStart)
 			}
 		}
 	})
@@ -628,6 +585,64 @@ func (l *Listener) Stream(ctx context.Context) error {
 	return group.Wait()
 }
 
+// handlePublishResult processes a single publish result delivered via the
+// result chan: it records monitor/log output for the publish outcome, returns
+// the event to the pool, and advances latestWalStart from any attached
+// keepalive or XLogData. A nil result is treated as a successful no-op
+// (used by oversized-message drops); LSN bookkeeping still runs so the slot
+// advances.
+func (l *Listener) handlePublishResult(
+	ctx context.Context,
+	object *eventAndPublishResult,
+	pool *sync.Pool,
+	latestWalStart *atomic.Uint64,
+) {
+	subjectName, result, event, xld, pkm := object.subjectName, object.result, object.event, object.xld, object.pkm
+
+	if result != nil {
+		_, err := result.Get(ctx)
+		if err != nil {
+			// Only warn on publish failures, but continue to process and ack messages
+			// this runs the risk of losing data, but this it is more important to keep processing WAL messages in the meantime
+			l.monitor.IncProblematicEvents(problemKindPublish)
+
+			l.log.Warn(
+				"failed to publish message",
+				slog.Any("error", err),
+				slog.String("subjectName", subjectName),
+				slog.String("table", event.Table),
+				slog.String("action", event.Action),
+			)
+		} else {
+			l.monitor.IncPublishedEvents(subjectName, event.Table)
+			l.log.Debug(
+				"event was sent",
+				slog.String("subject", subjectName),
+				slog.String("action", event.Action),
+				slog.String("schema", event.Schema),
+				slog.String("table", event.Table),
+				slog.String("lsn", l.readLSN().String()),
+			)
+		}
+		pool.Put(event)
+	}
+
+	latest := latestWalStart.Load()
+	if pkm != nil {
+		walEnd := uint64(pkm.ServerWALEnd)
+		if walEnd > latest {
+			latestWalStart.Store(walEnd)
+		}
+	} else if xld != nil {
+		// We do not need to compare and swap here as there's only one thread
+		// writing to this value
+		walStart := uint64(xld.WALStart)
+		if xld.WALData != nil && walStart > latest {
+			latestWalStart.Store(walStart)
+		}
+	}
+}
+
 func (l *Listener) processHeartBeat(ctx context.Context, pkm *pglogrepl.PrimaryKeepaliveMessage) {
 	if pkm.ReplyRequested {
 		l.log.Debug("status requested")

listener/listener_test.go

@@ -7,15 +7,19 @@ import (
 	"io"
 	"log/slog"
 	"strings"
+	"sync"
+	"sync/atomic"
 	"testing"
 	"time"
 
 	"github.com/ihippik/wal-listener/v2/config"
+	"github.com/ihippik/wal-listener/v2/publisher"
 	"github.com/jackc/pglogrepl"
 	"github.com/jackc/pgx/v5"
 	"github.com/jackc/pgx/v5/pgproto3"
 	"github.com/stretchr/testify/assert"
 	"github.com/stretchr/testify/mock"
+	"github.com/stretchr/testify/require"
 )
 
 var (
@@ -711,3 +715,99 @@ func TestListener_logRetainedWalBytes(t *testing.T) {
 		})
 	}
 }
+
+// countingMonitor is a monitor implementation that records call counts for assertions.
+type countingMonitor struct {
+	publishedEvents    int
+	filterSkippedEvts  int
+	problematicEvents  map[string]int
+	lastPublishSubject string
+	lastPublishTable   string
+}
+
+func newCountingMonitor() *countingMonitor {
+	return &countingMonitor{problematicEvents: make(map[string]int)}
+}
+
+func (m *countingMonitor) IncPublishedEvents(subject, table string) {
+	m.publishedEvents++
+	m.lastPublishSubject = subject
+	m.lastPublishTable = table
+}
+func (m *countingMonitor) IncFilterSkippedEvents(table string) { m.filterSkippedEvts++ }
+func (m *countingMonitor) IncProblematicEvents(kind string)    { m.problematicEvents[kind]++ }
+
+// TestListener_handlePublishResult_NilErrorResult verifies that when a publisher
+// returns publisher.NewPublishResult(nil) — as GooglePubSubPublisher does for
+// oversized messages it deliberately drops — the listener's result handler treats
+// it as a successful publish: no error metric, event returned to the pool, and
+// the attached XLogData advances latestWalStart.
+func TestListener_handlePublishResult_NilErrorResult(t *testing.T) {
+	logger := slog.New(slog.NewJSONHandler(io.Discard, nil))
+	monitor := newCountingMonitor()
+
+	l := &Listener{
+		log:     logger,
+		monitor: monitor,
+		cfg:     &config.Config{Listener: &config.ListenerCfg{}},
+	}
+
+	pool := &sync.Pool{New: func() any { return &publisher.Event{} }}
+	var latest atomic.Uint64
+
+	event := pool.Get().(*publisher.Event)
+	event.Schema = "public"
+	event.Table = "widgets"
+	event.Action = "update"
+
+	xldWALStart := pglogrepl.LSN(5000)
+	object := &eventAndPublishResult{
+		subjectName: "test.public_widgets",
+		event:       event,
+		result:      publisher.NewPublishResult(nil),
+		xld: &pglogrepl.XLogData{
+			WALStart: xldWALStart,
+			WALData:  []byte{0x01},
+		},
+	}
+
+	l.handlePublishResult(context.Background(), object, pool, &latest)
+
+	assert.Equal(t, 1, monitor.publishedEvents, "expected one successful publish increment")
+	assert.Equal(t, 0, monitor.problematicEvents[problemKindPublish], "expected no publish problem increment")
+	assert.Equal(t, "test.public_widgets", monitor.lastPublishSubject)
+	assert.Equal(t, "widgets", monitor.lastPublishTable)
+	assert.Equal(t, uint64(xldWALStart), latest.Load(), "expected latestWalStart to advance to xld.WALStart")
+
+	// Event should have been returned to the pool; fetching it back should yield the same pointer.
+	reused := pool.Get().(*publisher.Event)
+	require.Same(t, event, reused, "expected the event to be returned to the pool after publish")
+}
+
+// TestListener_handlePublishResult_NilResult verifies that an eventAndPublishResult
+// carrying a nil PublishResult (e.g. the "empty events" case) does not touch
+// monitor counters but still advances latestWalStart from any attached xld/pkm.
+func TestListener_handlePublishResult_NilResult(t *testing.T) {
+	logger := slog.New(slog.NewJSONHandler(io.Discard, nil))
+	monitor := newCountingMonitor()
+
+	l := &Listener{
+		log:     logger,
+		monitor: monitor,
+		cfg:     &config.Config{Listener: &config.ListenerCfg{}},
+	}
+
+	pool := &sync.Pool{New: func() any { return &publisher.Event{} }}
+	var latest atomic.Uint64
+
+	object := &eventAndPublishResult{
+		result: nil,
+		pkm:    &pglogrepl.PrimaryKeepaliveMessage{ServerWALEnd: pglogrepl.LSN(9000)},
+	}
+
+	l.handlePublishResult(context.Background(), object, pool, &latest)
+
+	assert.Equal(t, 0, monitor.publishedEvents)
+	assert.Equal(t, 0, monitor.problematicEvents[problemKindPublish])
+	assert.Equal(t, uint64(9000), latest.Load(), "expected latestWalStart to advance to pkm.ServerWALEnd")
+}

publisher/pubsub.go

@@ -6,6 +6,13 @@ import (
 	"fmt"
 )
 
+// Google Pub/Sub hard per-message limit is 10 MiB. The publish RPC also carries
+// protobuf framing, attributes, and the ordering key, so keep safely under.
+// Oversized events are dropped with a warning rather than passed to the client
+// bundler (which rejects them with "item size exceeds bundle byte limit" and
+// produces confusing, hard-to-action logs).
+const maxPubSubMessageBytes = 9 * 1024 * 1024
+
 // GooglePubSubPublisher represent Pub/Sub publisher.
 type GooglePubSubPublisher struct {
 	pubSubConnection *PubSubConnection
@@ -25,6 +32,19 @@ func (p *GooglePubSubPublisher) Publish(ctx context.Context, topic string, event
 		return NewPublishResult(fmt.Errorf("marshal: %w", err))
 	}
 
+	if len(body) > maxPubSubMessageBytes {
+		p.pubSubConnection.logger.Warn("dropping oversized message",
+			"topic", topic,
+			"schema", event.Schema,
+			"table", event.Table,
+			"action", event.Action,
+			"event_id", event.ID.String(),
+			"size_bytes", len(body),
+			"limit_bytes", maxPubSubMessageBytes,
+		)
+		return NewPublishResult(nil)
+	}
+
 	return p.pubSubConnection.Publish(ctx, topic, body)
 }
 

publisher/pubsub_test.go

@@ -0,0 +1,83 @@
+package publisher
+
+import (
+	"bytes"
+	"context"
+	"log/slog"
+	"strings"
+	"testing"
+
+	"github.com/google/uuid"
+)
+
+func TestGooglePubSubPublisher_DropsOversizedMessage(t *testing.T) {
+	var logBuf bytes.Buffer
+	conn := &PubSubConnection{
+		logger: slog.New(slog.NewTextHandler(&logBuf, &slog.HandlerOptions{Level: slog.LevelWarn})),
+	}
+	p := NewGooglePubSubPublisher(conn)
+
+	// A string payload larger than the limit guarantees the marshaled event
+	// exceeds maxPubSubMessageBytes.
+	big := strings.Repeat("x", maxPubSubMessageBytes+1)
+	eventID := uuid.New()
+	event := &Event{
+		ID:     eventID,
+		Schema: "public",
+		Table:  "big_table",
+		Action: "update",
+		Data:   map[string]any{"blob": big},
+	}
+
+	result := p.Publish(context.Background(), "some-topic", event)
+
+	serverID, err := result.Get(context.Background())
+	if err != nil {
+		t.Fatalf("expected nil error for dropped oversized event, got %v", err)
+	}
+	if serverID == "" {
+		t.Fatalf("expected non-empty serverID placeholder, got empty string")
+	}
+
+	logs := logBuf.String()
+	if !strings.Contains(logs, "dropping oversized message") {
+		t.Errorf("expected warning log, got: %q", logs)
+	}
+	if !strings.Contains(logs, "big_table") {
+		t.Errorf("expected table name in log, got: %q", logs)
+	}
+	if !strings.Contains(logs, eventID.String()) {
+		t.Errorf("expected event id in log, got: %q", logs)
+	}
+}
+
+func TestGooglePubSubPublisher_UnderLimitIsNotDropped(t *testing.T) {
+	// A small event should not take the drop path. We can't safely exercise the
+	// downstream publish path here (it requires a real pubsub client), so we
+	// verify the size check by confirming the drop-path warning is NOT emitted
+	// for an under-limit event. The test then recovers from the expected nil
+	// client panic when the real publish is attempted.
+	var logBuf bytes.Buffer
+	conn := &PubSubConnection{
+		logger: slog.New(slog.NewTextHandler(&logBuf, &slog.HandlerOptions{Level: slog.LevelWarn})),
+	}
+	p := NewGooglePubSubPublisher(conn)
+
+	event := &Event{
+		ID:     uuid.New(),
+		Schema: "public",
+		Table:  "small_table",
+		Action: "insert",
+		Data:   map[string]any{"v": "hello"},
+	}
+
+	defer func() {
+		// Recover from the expected panic when Publish hits the nil pubsub client.
+		_ = recover()
+		if strings.Contains(logBuf.String(), "dropping oversized message") {
+			t.Errorf("did not expect drop-path warning for under-limit event, got logs: %q", logBuf.String())
+		}
+	}()
+
+	_ = p.Publish(context.Background(), "some-topic", event)
+}