From 061e3ba3acd045b2ff8d13d8a8a9e20f85918f90 Mon Sep 17 00:00:00 2001
From: Peter Stuifzand <peter@madebymouses.com>
Date: Sun, 8 Nov 2026 01:42:00 +0100
Subject: [PATCH] Close the rows an incoming publish opens

UpdateFeed and Feeds query and never close the result set, so every WebSub
delivery holds a pooled connection until the process exits. The database hit
max_connections tonight, psql could not connect, postgresStream.Init() failed,
timeline.Create returned nil and every push answered 400 "timeline not found"
- with an error that named a timeline rather than the exhausted pool. The
reader had received no pushes at all.

The feed body is also read once now: passing the request body straight to
ProcessContent empties it for the second channel a feed belongs to.

Tests need the redis and postgres services the pipeline starts; the build was
checked locally, not the suite.
---
 cmd/eksterd/hubbackend.go | 19 ++++++++++++++++---
 1 file changed, 16 insertions(+), 3 deletions(-)

diff --git a/cmd/eksterd/hubbackend.go b/cmd/eksterd/hubbackend.go
index 102628b..4912b42 100644
--- a/cmd/eksterd/hubbackend.go
+++ b/cmd/eksterd/hubbackend.go
@@ -19,6 +19,7 @@
 package main
 
 import (
+	"bytes"
 	"database/sql"
 	"expvar"
 	"fmt"
@@ -135,6 +136,13 @@ func (h *hubIncomingBackend) UpdateFeed(processor ContentProcessor, subscription
 	log.Println("UpdateFeed", subscriptionID)
 
 	db := h.database
+	// Read the body once. Passing the reader straight to ProcessContent
+	// empties it for the second channel a feed belongs to.
+	content, err := io.ReadAll(body)
+	if err != nil {
+		return err
+	}
+
 	// Process all channels that contains this feed
 	rows, err := db.Query(`
 select topic, c.uid, f.id, c.name
@@ -148,6 +156,10 @@ where s.id = $1
 	if err != nil {
 		return err
 	}
+	// A rows left open holds a pooled connection until the process exits; one
+	// per delivery exhausts the database's max_connections, and every later
+	// query fails with "too many clients already".
+	defer rows.Close()
 
 	for rows.Next() {
 		var topic, channel, feedID, channelName string
@@ -159,13 +171,13 @@ where s.id = $1
 		}
 
 		log.Printf("Updating feed %s %q in %q (%s)\n", feedID, topic, channelName, channel)
-		_, err = processor.ProcessContent(channel, feedID, topic, contentType, body)
+		_, err = processor.ProcessContent(channel, feedID, topic, contentType, bytes.NewReader(content))
 		if err != nil {
 			log.Printf("could not process content for channel %s: %s", channelName, err)
 		}
 	}
 
-	return err
+	return rows.Err()
 }
 
 func (h *hubIncomingBackend) FeedSetLeaseSeconds(subscriptionID int64, leaseSeconds int64) error {
@@ -194,6 +206,7 @@ func (h *hubIncomingBackend) Feeds() ([]Feed, error) {
 	if err != nil {
 		return nil, err
 	}
+	defer rows.Close()
 
 	for rows.Next() {
 		var feed Feed
@@ -214,7 +227,7 @@ func (h *hubIncomingBackend) Feeds() ([]Feed, error) {
 		feeds = append(feeds, feed)
 	}
 
-	return feeds, nil
+	return feeds, rows.Err()
 }
 
 func (h *hubIncomingBackend) Subscribe(feed *Feed) error {