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.
This commit is contained in:
Peter Stuifzand 2026-11-08 01:42:00 +01:00
parent edf5a029ee
commit 061e3ba3ac
No known key found for this signature in database

View File

@ -19,6 +19,7 @@
package main package main
import ( import (
"bytes"
"database/sql" "database/sql"
"expvar" "expvar"
"fmt" "fmt"
@ -135,6 +136,13 @@ func (h *hubIncomingBackend) UpdateFeed(processor ContentProcessor, subscription
log.Println("UpdateFeed", subscriptionID) log.Println("UpdateFeed", subscriptionID)
db := h.database 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 // Process all channels that contains this feed
rows, err := db.Query(` rows, err := db.Query(`
select topic, c.uid, f.id, c.name select topic, c.uid, f.id, c.name
@ -148,6 +156,10 @@ where s.id = $1
if err != nil { if err != nil {
return err 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() { for rows.Next() {
var topic, channel, feedID, channelName string 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) 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 { if err != nil {
log.Printf("could not process content for channel %s: %s", channelName, err) 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 { func (h *hubIncomingBackend) FeedSetLeaseSeconds(subscriptionID int64, leaseSeconds int64) error {
@ -194,6 +206,7 @@ func (h *hubIncomingBackend) Feeds() ([]Feed, error) {
if err != nil { if err != nil {
return nil, err return nil, err
} }
defer rows.Close()
for rows.Next() { for rows.Next() {
var feed Feed var feed Feed
@ -214,7 +227,7 @@ func (h *hubIncomingBackend) Feeds() ([]Feed, error) {
feeds = append(feeds, feed) feeds = append(feeds, feed)
} }
return feeds, nil return feeds, rows.Err()
} }
func (h *hubIncomingBackend) Subscribe(feed *Feed) error { func (h *hubIncomingBackend) Subscribe(feed *Feed) error {