ai/live: Reuse HTTP connections for trickle requests (#3470)

Initially we were re-creating connections to avoid being stuck
in a degraded state over time, but it seems that we may have
trouble even establishing a connection sometimes. Attempt to
re-use established connections to mitigate this problem.

Both scenarios should be caught by the slow orchestrator checker,
but hopefully connection reuse makes the checker trip less often.

Author: j0sh

trickle/trickle_publisher.go

@@ -14,6 +14,7 @@ var StreamNotFoundErr = errors.New("stream not found")
 
 // TricklePublisher represents a trickle streaming client
 type TricklePublisher struct {
+	client      *http.Client
 	baseURL     string
 	index       int          // Current index for segments
 	writeLock   sync.Mutex   // Mutex to manage concurrent access
@@ -47,6 +48,7 @@ func NewTricklePublisher(url string) (*TricklePublisher, error) {
 	c := &TricklePublisher{
 		baseURL:     url,
 		contentType: "video/MP2T",
+		client:      httpClient(),
 	}
 	p, err := c.preconnect()
 	if err != nil {
@@ -76,13 +78,7 @@ func (c *TricklePublisher) preconnect() (*pendingPost, error) {
 
 	// Start the POST request in a background goroutine
 	go func() {
-		// Createa new client to prevent connection reuse
-		client := http.Client{Transport: &http.Transport{
-			DisableKeepAlives: true,
-			// ignore orch certs for now
-			TLSClientConfig: &tls.Config{InsecureSkipVerify: true},
-		}}
-		resp, err := client.Do(req)
+		resp, err := c.client.Do(req)
 		if err != nil {
 			slog.Error("Failed to complete POST for segment", "url", url, "err", err)
 			errCh <- err
@@ -126,11 +122,8 @@ func (c *TricklePublisher) Close() error {
 	if err != nil {
 		return err
 	}
-	resp, err := (&http.Client{Transport: &http.Transport{
-		DisableKeepAlives: true,
-		// ignore orch certs for now
-		TLSClientConfig: &tls.Config{InsecureSkipVerify: true},
-	}}).Do(req)
+	// Use a new client for a fresh connection
+	resp, err := httpClient().Do(req)
 	if err != nil {
 		return err
 	}
@@ -174,11 +167,12 @@ func (p *pendingPost) reconnect() (*pendingPost, error) {
 	// Set the publisher's sequence sequence to the intended reconnect
 	// Call publisher's preconnect (which increments its sequence)
 	// then reset publisher's sequence back to the original
-	//slog.Info("Re-connecting", "url", p.client.baseURL, "seq", p.client.index)
+	// Also recreate the client to force a fresh connection
 	p.client.writeLock.Lock()
 	defer p.client.writeLock.Unlock()
 	currentSeq := p.client.index
 	p.client.index = p.index
+	p.client.client = httpClient()
 	pp, err := p.client.preconnect()
 	p.client.index = currentSeq
 	return pp, err
@@ -261,11 +255,10 @@ func (p *pendingPost) Close() error {
 	if err != nil {
 		return err
 	}
-	resp, err := (&http.Client{Transport: &http.Transport{
-		DisableKeepAlives: true,
-		// ignore orch certs for now
-		TLSClientConfig: &tls.Config{InsecureSkipVerify: true},
-	}}).Do(req)
+	// Since this method typically gets invoked when
+	// there is a problem sending the segment, use a
+	// new client for a fresh connection just in case
+	resp, err := httpClient().Do(req)
 	if err != nil {
 		return err
 	}
@@ -287,6 +280,15 @@ func (c *TricklePublisher) Write(data io.Reader) error {
 	return err
 }
 
+func httpClient() *http.Client {
+	return &http.Client{Transport: &http.Transport{
+		// Re-enable keepalives to avoid connection pooling
+		// DisableKeepAlives: true,
+		// ignore orch certs for now
+		TLSClientConfig: &tls.Config{InsecureSkipVerify: true},
+	}}
+}
+
 func humanBytes(bytes int64) string {
 	var unit int64 = 1024
 	if bytes < unit {

trickle/trickle_subscriber.go

@@ -2,7 +2,6 @@ package trickle
 
 import (
 	"context"
-	"crypto/tls"
 	"errors"
 	"fmt"
 	"io"
@@ -28,6 +27,7 @@ const preconnectRefreshTimeout = 20 * time.Second
 
 // TrickleSubscriber represents a trickle streaming reader that always fetches from index -1
 type TrickleSubscriber struct {
+	client     *http.Client
 	url        string
 	mu         sync.Mutex     // Mutex to manage concurrent access
 	pendingGet *http.Response // Pre-initialized GET request
@@ -41,8 +41,9 @@ type TrickleSubscriber struct {
 func NewTrickleSubscriber(url string) *TrickleSubscriber {
 	// No preconnect needed here; it will be handled by the first Read call.
 	return &TrickleSubscriber{
-		url: url,
-		idx: -1, // shortcut for 'latest'
+		client: httpClient(),
+		url:    url,
+		idx:    -1, // shortcut for 'latest'
 	}
 }
 
@@ -93,10 +94,7 @@ func (c *TrickleSubscriber) connect(ctx context.Context) (*http.Response, error)
 	}
 
 	// Execute the GET request
-	resp, err := (&http.Client{Transport: &http.Transport{
-		DisableKeepAlives: true,
-		TLSClientConfig:   &tls.Config{InsecureSkipVerify: true},
-	}}).Do(req)
+	resp, err := c.client.Do(req)
 	if err != nil {
 		return nil, fmt.Errorf("failed to complete GET for next segment: %w", err)
 	}