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 {