Keep the message batch writer ahead of publish bursts

On cron bursts (top of the hour), the single batch writer fell behind: it wrote one INSERT round
trip per message, so against a busy remote Postgres it managed ~100 rows/s while ~300/s arrived.
New messages took up to ~40s to reach the database (polls missed them), and publishes blocked on
the unbuffered queue for up to ~25s.

- Postgres: write each batch with one multi-row INSERT (up to 1000 rows per statement) instead of
  one statement per row. SQLite keeps the row-by-row path.
- Skip DeleteScheduledBySequenceID when the client did not set a sequence ID: the sequence ID is
  then the message's own fresh ID, so no scheduled message can match. This saves a transaction on
  the primary for almost every publish.
- Buffer up to 100 batches in the BatchingQueue, so a briefly slow writer does not stall publishes.
- dbtest: add a latency proxy (CreateTestPostgresWithLatency, WithLatency) to test round trips.

db/test/test.go

@@ -2,9 +2,12 @@ package dbtest
 
 import (
 	"fmt"
+	"io"
+	"net"
 	"net/url"
 	"os"
 	"testing"
+	"time"
 
 	"github.com/stretchr/testify/require"
 	"heckel.io/ntfy/v2/db"
@@ -62,3 +65,71 @@ func CreateTestPostgres(t *testing.T) *db.DB {
 	})
 	return d
 }
+
+// CreateTestPostgresWithLatency is CreateTestPostgres, but every request the client sends reaches
+// the database latency later, so each database round trip costs at least latency (as it does
+// against a remote managed database). Use it to test how many round trips a code path makes.
+func CreateTestPostgresWithLatency(t *testing.T, latency time.Duration) *db.DB {
+	t.Helper()
+	testHost, err := pg.Open(WithLatency(t, CreateTestPostgresSchema(t), latency))
+	require.Nil(t, err)
+	d := db.New(testHost, nil)
+	t.Cleanup(func() {
+		d.Close()
+	})
+	return d
+}
+
+// WithLatency starts a local TCP proxy to the database in dsn that delays everything the client
+// sends by latency, and returns dsn pointing at the proxy. The proxy stops when the test ends.
+func WithLatency(t *testing.T, dsn string, latency time.Duration) string {
+	t.Helper()
+	u, err := url.Parse(dsn)
+	require.Nil(t, err)
+	upstream := u.Host
+	if u.Port() == "" {
+		upstream = net.JoinHostPort(u.Hostname(), "5432")
+	}
+	listener, err := net.Listen("tcp", "127.0.0.1:0")
+	require.Nil(t, err)
+	t.Cleanup(func() {
+		listener.Close()
+	})
+	go func() {
+		for {
+			client, err := listener.Accept()
+			if err != nil {
+				return // Listener closed
+			}
+			go proxyWithLatency(client, upstream, latency)
+		}
+	}()
+	u.Host = listener.Addr().String()
+	return u.String()
+}
+
+func proxyWithLatency(client net.Conn, upstream string, latency time.Duration) {
+	defer client.Close()
+	server, err := net.Dial("tcp", upstream)
+	if err != nil {
+		return
+	}
+	defer server.Close()
+	go func() {
+		io.Copy(client, server)
+		client.Close()
+	}()
+	buf := make([]byte, 64*1024)
+	for {
+		n, err := client.Read(buf)
+		if n > 0 {
+			time.Sleep(latency)
+			if _, err := server.Write(buf[:n]); err != nil {
+				return
+			}
+		}
+		if err != nil {
+			return
+		}
+	}
+}

message/cache.go

@@ -27,6 +27,13 @@ const (
 	// write takes milliseconds, so this only ever trips when the database is wedged, and then
 	// shutdown must still finish well inside systemd's stop timeout.
 	closeFlushTimeout = 5 * time.Second
+
+	// insertMessageColumns is the number of values per row in insertMessage and insertMessages
+	insertMessageColumns = 24
+
+	// insertMessagesMaxRows caps the rows per multi-row INSERT, keeping it well below PostgreSQL's
+	// limit of 65535 parameters per statement
+	insertMessagesMaxRows = 1000
 )
 
 var errNoRows = errors.New("no rows found")
@@ -34,6 +41,7 @@ var errNoRows = errors.New("no rows found")
 // queries holds the database-specific SQL queries
 type queries struct {
 	insertMessage                    string
+	insertMessages                   func(rows int) string // Multi-row INSERT; nil writes row by row with insertMessage
 	selectScheduledMessageIDsBySeqID string
 	deleteScheduledBySequenceID      string
 	updateMessagesForTopicExpiry     string
@@ -127,69 +135,14 @@ func (c *Cache) addMessages(ms []*model.Message) error {
 		return err
 	}
 	defer tx.Rollback()
-	stmt, err := tx.Prepare(c.queries.insertMessage)
+	if c.queries.insertMessages != nil {
+		err = insertMessagesMultiRow(tx, c.queries.insertMessages, ms)
+	} else {
+		err = insertMessagesRowByRow(tx, c.queries.insertMessage, ms)
+	}
 	if err != nil {
 		return err
 	}
-	defer stmt.Close()
-	for _, m := range ms {
-		if m.Event != model.MessageEvent && m.Event != model.MessageDeleteEvent && m.Event != model.MessageClearEvent {
-			return model.ErrUnexpectedMessageType
-		}
-		published := m.Time <= time.Now().Unix()
-		tags := util.SanitizeUTF8(strings.Join(m.Tags, ","))
-		var attachmentName, attachmentType, attachmentURL string
-		var attachmentSize, attachmentExpires int64
-		var attachmentDeleted bool
-		if m.Attachment != nil {
-			attachmentName = util.SanitizeUTF8(m.Attachment.Name)
-			attachmentType = util.SanitizeUTF8(m.Attachment.Type)
-			attachmentSize = m.Attachment.Size
-			attachmentExpires = m.Attachment.Expires
-			attachmentURL = util.SanitizeUTF8(m.Attachment.URL)
-		}
-		var actionsStr string
-		if len(m.Actions) > 0 {
-			actionsBytes, err := json.Marshal(m.Actions)
-			if err != nil {
-				return err
-			}
-			actionsStr = string(actionsBytes)
-		}
-		var sender string
-		if m.Sender.IsValid() {
-			sender = m.Sender.String()
-		}
-		_, err := stmt.Exec(
-			m.ID,
-			m.SequenceID,
-			m.Time,
-			m.Event,
-			m.Expires,
-			util.SanitizeUTF8(m.Topic),
-			util.SanitizeUTF8(m.Message),
-			util.SanitizeUTF8(m.Title),
-			m.Priority,
-			tags,
-			util.SanitizeUTF8(m.Click),
-			util.SanitizeUTF8(m.Icon),
-			actionsStr,
-			attachmentName,
-			attachmentType,
-			attachmentSize,
-			attachmentExpires,
-			attachmentURL,
-			attachmentDeleted, // Always zero
-			sender,
-			m.User,
-			util.SanitizeUTF8(m.ContentType),
-			m.Encoding,
-			published,
-		)
-		if err != nil {
-			return err
-		}
-	}
 	if err := tx.Commit(); err != nil {
 		log.Tag(tagMessageCache).Err(err).Error("Writing %d message(s) failed (took %v)", len(ms), time.Since(start))
 		return err
@@ -198,6 +151,102 @@ func (c *Cache) addMessages(ms []*model.Message) error {
 	return nil
 }
 
+// insertMessagesMultiRow writes the messages with as few statements as possible: against a remote
+// database, every statement is a round trip, and the batch writer is the only writer
+func insertMessagesMultiRow(tx *sql.Tx, query func(rows int) string, ms []*model.Message) error {
+	for len(ms) > 0 {
+		chunk := ms[:min(len(ms), insertMessagesMaxRows)]
+		ms = ms[len(chunk):]
+		args := make([]any, 0, len(chunk)*insertMessageColumns)
+		for _, m := range chunk {
+			rowArgs, err := insertMessageArgs(m)
+			if err != nil {
+				return err
+			}
+			args = append(args, rowArgs...)
+		}
+		if _, err := tx.Exec(query(len(chunk)), args...); err != nil {
+			return err
+		}
+	}
+	return nil
+}
+
+func insertMessagesRowByRow(tx *sql.Tx, query string, ms []*model.Message) error {
+	stmt, err := tx.Prepare(query)
+	if err != nil {
+		return err
+	}
+	defer stmt.Close()
+	for _, m := range ms {
+		args, err := insertMessageArgs(m)
+		if err != nil {
+			return err
+		}
+		if _, err := stmt.Exec(args...); err != nil {
+			return err
+		}
+	}
+	return nil
+}
+
+// insertMessageArgs returns the insertMessageColumns values of one message row, in column order
+func insertMessageArgs(m *model.Message) ([]any, error) {
+	if m.Event != model.MessageEvent && m.Event != model.MessageDeleteEvent && m.Event != model.MessageClearEvent {
+		return nil, model.ErrUnexpectedMessageType
+	}
+	published := m.Time <= time.Now().Unix()
+	tags := util.SanitizeUTF8(strings.Join(m.Tags, ","))
+	var attachmentName, attachmentType, attachmentURL string
+	var attachmentSize, attachmentExpires int64
+	var attachmentDeleted bool
+	if m.Attachment != nil {
+		attachmentName = util.SanitizeUTF8(m.Attachment.Name)
+		attachmentType = util.SanitizeUTF8(m.Attachment.Type)
+		attachmentSize = m.Attachment.Size
+		attachmentExpires = m.Attachment.Expires
+		attachmentURL = util.SanitizeUTF8(m.Attachment.URL)
+	}
+	var actionsStr string
+	if len(m.Actions) > 0 {
+		actionsBytes, err := json.Marshal(m.Actions)
+		if err != nil {
+			return nil, err
+		}
+		actionsStr = string(actionsBytes)
+	}
+	var sender string
+	if m.Sender.IsValid() {
+		sender = m.Sender.String()
+	}
+	return []any{
+		m.ID,
+		m.SequenceID,
+		m.Time,
+		m.Event,
+		m.Expires,
+		util.SanitizeUTF8(m.Topic),
+		util.SanitizeUTF8(m.Message),
+		util.SanitizeUTF8(m.Title),
+		m.Priority,
+		tags,
+		util.SanitizeUTF8(m.Click),
+		util.SanitizeUTF8(m.Icon),
+		actionsStr,
+		attachmentName,
+		attachmentType,
+		attachmentSize,
+		attachmentExpires,
+		attachmentURL,
+		attachmentDeleted, // Always zero
+		sender,
+		m.User,
+		util.SanitizeUTF8(m.ContentType),
+		m.Encoding,
+		published,
+	}, nil
+}
+
 // Messages returns all cached messages for a topic, oldest first. Prefer MessagesCapped on
 // request paths: an uncapped replay of a busy topic is as large as the topic's entire cache.
 func (c *Cache) Messages(topic string, since model.SinceMarker, scheduled bool) ([]*model.Message, error) {

message/cache_postgres.go

@@ -1,6 +1,8 @@
 package message
 
 import (
+	"strconv"
+	"strings"
 	"time"
 
 	"heckel.io/ntfy/v2/db"
@@ -9,10 +11,9 @@ import (
 
 // PostgreSQL runtime query constants
 const (
-	postgresInsertMessageQuery = `
+	postgresInsertMessagesQuery = `
 		INSERT INTO message (mid, sequence_id, time, event, expires, topic, message, title, priority, tags, click, icon, actions, attachment_name, attachment_type, attachment_size, attachment_expires, attachment_url, attachment_deleted, sender, user_id, content_type, encoding, published)
-		VALUES ($1, $2, $3, $4, $5, $6, $7, $8, $9, $10, $11, $12, $13, $14, $15, $16, $17, $18, $19, $20, $21, $22, $23, $24)
-	`
+		VALUES `
 	postgresSelectScheduledMessageIDsBySeqIDQuery = `SELECT mid FROM message WHERE topic = $1 AND sequence_id = $2 AND published = FALSE`
 	postgresDeleteScheduledBySequenceIDQuery      = `DELETE FROM message WHERE topic = $1 AND sequence_id = $2 AND published = FALSE`
 	postgresUpdateMessagesForTopicExpiryQuery     = `UPDATE message SET expires = $1 WHERE topic = $2`
@@ -83,7 +84,7 @@ const (
 )
 
 var postgresQueries = queries{
-	insertMessage:                    postgresInsertMessageQuery,
+	insertMessages:                   postgresInsertMessages,
 	selectScheduledMessageIDsBySeqID: postgresSelectScheduledMessageIDsBySeqIDQuery,
 	deleteScheduledBySequenceID:      postgresDeleteScheduledBySequenceIDQuery,
 	updateMessagesForTopicExpiry:     postgresUpdateMessagesForTopicExpiryQuery,
@@ -115,3 +116,23 @@ func NewPostgresStore(d *db.DB, batchSize int, batchTimeout time.Duration) (*Cac
 	}
 	return newCache(d, postgresQueries, nil, batchSize, batchTimeout, false), nil
 }
+
+// postgresInsertMessages returns a multi-row INSERT for the given number of message rows
+func postgresInsertMessages(rows int) string {
+	var b strings.Builder
+	b.WriteString(postgresInsertMessagesQuery)
+	for r := 0; r < rows; r++ {
+		if r > 0 {
+			b.WriteString(", ")
+		}
+		b.WriteString("(")
+		for c := 0; c < insertMessageColumns; c++ {
+			if c > 0 {
+				b.WriteString(", ")
+			}
+			b.WriteString("$" + strconv.Itoa(r*insertMessageColumns+c+1))
+		}
+		b.WriteString(")")
+	}
+	return b.String()
+}

message/cache_test.go

@@ -695,6 +695,78 @@ func TestStore_AddMessages(t *testing.T) {
 	})
 }
 
+func TestStore_AddMessages_AllFields(t *testing.T) {
+	forEachBackend(t, func(t *testing.T, s *message.Cache) {
+		// Every column must land in the right row when several messages are written in one batch
+		msgs := make([]*model.Message, 0)
+		for i := 0; i < 3; i++ {
+			m := model.NewDefaultMessage("mytopic", fmt.Sprintf("message %d", i))
+			m.Time = int64(1000 + i)
+			m.SequenceID = fmt.Sprintf("seq%d", i)
+			m.Title = fmt.Sprintf("title %d", i)
+			m.Priority = i + 1
+			m.Tags = []string{fmt.Sprintf("tag%d", i), "common"}
+			m.Click = fmt.Sprintf("https://example.com/%d", i)
+			m.Icon = fmt.Sprintf("https://example.com/%d.png", i)
+			m.Actions = []*model.Action{{ID: fmt.Sprintf("a%d", i), Action: "view", Label: fmt.Sprintf("label %d", i), URL: "https://example.com"}}
+			m.Attachment = &model.Attachment{Name: fmt.Sprintf("file%d.txt", i), Type: "text/plain", Size: int64(10 + i), Expires: int64(2000 + i), URL: fmt.Sprintf("https://example.com/file%d.txt", i)}
+			m.Sender = netip.MustParseAddr(fmt.Sprintf("1.2.3.%d", i+1))
+			m.User = fmt.Sprintf("u_%d", i)
+			m.ContentType = "text/markdown"
+			m.Encoding = "base64"
+			msgs = append(msgs, m)
+		}
+		require.Nil(t, s.AddMessages(msgs))
+
+		messages, err := s.Messages("mytopic", model.SinceAllMessages, false)
+		require.Nil(t, err)
+		require.Equal(t, 3, len(messages))
+		for i, m := range messages {
+			require.Equal(t, msgs[i].ID, m.ID)
+			require.Equal(t, fmt.Sprintf("seq%d", i), m.SequenceID)
+			require.Equal(t, int64(1000+i), m.Time)
+			require.Equal(t, fmt.Sprintf("message %d", i), m.Message)
+			require.Equal(t, fmt.Sprintf("title %d", i), m.Title)
+			require.Equal(t, i+1, m.Priority)
+			require.Equal(t, []string{fmt.Sprintf("tag%d", i), "common"}, m.Tags)
+			require.Equal(t, fmt.Sprintf("https://example.com/%d", i), m.Click)
+			require.Equal(t, fmt.Sprintf("https://example.com/%d.png", i), m.Icon)
+			require.Equal(t, 1, len(m.Actions))
+			require.Equal(t, fmt.Sprintf("label %d", i), m.Actions[0].Label)
+			require.Equal(t, fmt.Sprintf("file%d.txt", i), m.Attachment.Name)
+			require.Equal(t, int64(10+i), m.Attachment.Size)
+			require.Equal(t, int64(2000+i), m.Attachment.Expires)
+			require.Equal(t, fmt.Sprintf("https://example.com/file%d.txt", i), m.Attachment.URL)
+			require.Equal(t, fmt.Sprintf("1.2.3.%d", i+1), m.Sender.String())
+			require.Equal(t, fmt.Sprintf("u_%d", i), m.User)
+			require.Equal(t, "text/markdown", m.ContentType)
+			require.Equal(t, "base64", m.Encoding)
+		}
+	})
+}
+
+func TestPostgresStore_AddMessages_OneRoundTripPerBatch(t *testing.T) {
+	// The batch writer is the only writer, so its speed caps the publish rate on cron bursts. With
+	// one INSERT per row, a batch of 100 against a remote database costs 100+ round trips.
+	const latency = 10 * time.Millisecond
+	testDB := dbtest.CreateTestPostgresWithLatency(t, latency)
+	s, err := message.NewPostgresStore(testDB, 0, 0)
+	require.Nil(t, err)
+	msgs := make([]*model.Message, 0)
+	for i := 0; i < 100; i++ {
+		msgs = append(msgs, model.NewDefaultMessage("mytopic", fmt.Sprintf("message %d", i)))
+	}
+	require.Nil(t, s.AddMessages(msgs[:1])) // Warm up the pool connection
+	start := time.Now()
+	require.Nil(t, s.AddMessages(msgs[1:]))
+	took := time.Since(start)
+	require.Less(t, took, 10*latency, "writing 99 messages took %v", took)
+
+	messages, err := s.Messages("mytopic", model.SinceAllMessages, false)
+	require.Nil(t, err)
+	require.Equal(t, 100, len(messages))
+}
+
 func TestStore_MessagesDue(t *testing.T) {
 	forEachBackend(t, func(t *testing.T, s *message.Cache) {
 		// Add a message scheduled in the past (i.e. it's due now)

server/server.go

@@ -989,15 +989,18 @@ func (s *Server) handlePublishInternal(r *http.Request, v *visitor) (*model.Mess
 		logvrm(v, r, m).Tag(tagPublish).Debug("Message delayed, will process later")
 	}
 	if cache {
-		// Delete any existing scheduled message with the same sequence ID
-		deletedIDs, err := s.messageCache.DeleteScheduledBySequenceID(t.ID, m.SequenceID)
-		if err != nil {
-			return nil, err
-		}
-		// Delete attachment files for deleted scheduled messages
-		if s.attachment != nil && len(deletedIDs) > 0 {
-			if err := s.attachment.Remove(deletedIDs...); err != nil {
-				logvrm(v, r, m).Tag(tagPublish).Err(err).Warn("Error removing attachments for deleted scheduled messages")
+		// Delete any existing scheduled message with the same sequence ID. Without a client-provided
+		// sequence ID, it is the message's own fresh ID and nothing can match, so skip the round trip.
+		if m.SequenceID != m.ID {
+			deletedIDs, err := s.messageCache.DeleteScheduledBySequenceID(t.ID, m.SequenceID)
+			if err != nil {
+				return nil, err
+			}
+			// Delete attachment files for deleted scheduled messages
+			if s.attachment != nil && len(deletedIDs) > 0 {
+				if err := s.attachment.Remove(deletedIDs...); err != nil {
+					logvrm(v, r, m).Tag(tagPublish).Err(err).Warn("Error removing attachments for deleted scheduled messages")
+				}
 			}
 		}
 		logvrm(v, r, m).Tag(tagPublish).Debug("Adding message to cache")

server/server_test.go

@@ -4906,6 +4906,31 @@ func TestServer_UpdateScheduledMessage(t *testing.T) {
 	})
 }
 
+func TestServer_PublishWithoutSequenceID_NoDatabaseRoundTrip(t *testing.T) {
+	// A message without a client-provided sequence ID gets its own fresh ID as sequence ID, so no
+	// scheduled message can share it; publishing it must not wait for the database. Each round
+	// trip costs latency here, like against a remote database.
+	const latency = 50 * time.Millisecond
+	conf := newTestConfig(t, dbtest.WithLatency(t, dbtest.CreateTestPostgresSchema(t), latency))
+	conf.CacheBatchSize = 10
+	conf.CacheBatchTimeout = 100 * time.Millisecond
+	conf.AuthAccessCacheEnabled = true // Like on high-volume servers; otherwise the ACL check is a round trip
+	s := newTestServer(t, conf)
+	request(t, s, "PUT", "/mytopic", "warm up", nil)
+
+	start := time.Now()
+	response := request(t, s, "PUT", "/mytopic", "hi", nil)
+	took := time.Since(start)
+	require.Equal(t, 200, response.Code)
+	require.Less(t, took, latency, "publish took %v", took)
+
+	// The message still reaches the database
+	require.Eventually(t, func() bool {
+		response := request(t, s, "GET", "/mytopic/json?poll=1", "", nil)
+		return strings.Contains(response.Body.String(), `"message":"hi"`)
+	}, 5*time.Second, 100*time.Millisecond)
+}
+
 func TestServer_DeleteScheduledMessage(t *testing.T) {
 	forEachBackend(t, func(t *testing.T, databaseURL string) {
 		t.Parallel()

util/batching_queue.go

@@ -5,6 +5,13 @@ import (
 	"time"
 )
 
+const (
+	// batchingQueueBufferedBatches is how many full batches can wait for the consumer before
+	// Enqueue blocks. Callers of Enqueue are request handlers (publishes), so a briefly slow
+	// consumer must not stall them; a consumer that stays slow still pushes back eventually.
+	batchingQueueBufferedBatches = 100
+)
+
 // BatchingQueue is a queue that creates batches of the enqueued elements based on a
 // max batch size and a batch timeout.
 //
@@ -40,7 +47,7 @@ func NewBatchingQueue[T any](batchSize int, timeout time.Duration) *BatchingQueu
 		batchSize: batchSize,
 		timeout:   timeout,
 		in:        make([]T, 0),
-		out:       make(chan []T),
+		out:       make(chan []T, batchingQueueBufferedBatches),
 		done:      make(chan struct{}),
 	}
 	go q.timeoutTicker()

util/batching_queue_test.go

@@ -81,3 +81,26 @@ func TestBatchingQueue_CloseFlushesRemaining(t *testing.T) {
 	}
 	q.Enqueue(4) // Must not panic after Close
 }
+
+func TestBatchingQueue_EnqueueDoesNotBlockOnBusyConsumer(t *testing.T) {
+	// A slow consumer (e.g. the message batch writer on a busy database) must not stall the
+	// goroutines that enqueue (e.g. publish requests) while there is room in the buffer
+	q := util.NewBatchingQueue[int](2, time.Hour)
+	done := make(chan struct{})
+	go func() {
+		for i := 0; i < 20; i++ { // 10 full batches, nobody reading yet
+			q.Enqueue(i)
+		}
+		close(done)
+	}()
+	select {
+	case <-done:
+	case <-time.After(2 * time.Second):
+		t.Fatal("Enqueue blocked while the consumer was busy")
+	}
+	total := 0
+	for i := 0; i < 10; i++ {
+		total += len(<-q.Dequeue())
+	}
+	require.Equal(t, 20, total)
+}