fix(activitypub): fixes #4372 where excessive AP resource use disrupts video delivery (#4552)

activitypub/activitypub.go

@@ -1,8 +1,6 @@
 package activitypub
 
 import (
-	"math"
-
 	"github.com/owncast/owncast/activitypub/crypto"
 	"github.com/owncast/owncast/activitypub/inbox"
 	"github.com/owncast/owncast/activitypub/outbox"
@@ -36,14 +34,32 @@ func Start(datastore *data.Datastore) {
 }
 
 func getOutboundWorkerPoolSize() int {
+	// Use a reasonable fixed worker pool size instead of scaling with followers
+	// This prevents excessive resource usage when streamers have many followers
+	const (
+		minWorkers     = 10 // Minimum workers for small instances
+		maxWorkers     = 50 // Maximum workers to prevent resource exhaustion
+		defaultWorkers = 20 // Default for most instances
+	)
+
 	var followerCount int64
 	fc, err := persistence.GetFollowerCount()
 	if err != nil {
 		log.Errorln("Unable to get follower count", err)
-		fc = 50 // Arbitrary fallback value.
+		return defaultWorkers
 	}
-	followerCount = int64(math.Max(float64(fc), 50))
+	followerCount = fc
-	return int(followerCount * 5)
+
+	// Scale more conservatively: start with base workers, add 1 worker per 100 followers
+	// This gives a much more reasonable scaling than the previous followerCount * 5
+	workers := minWorkers + int(followerCount/100)
+
+	if workers > maxWorkers {
+		workers = maxWorkers
+	}
+
+	log.Infof("Initializing ActivityPub outbound worker pool with %d workers for %d followers", workers, followerCount)
+	return workers
 }
 
 // SendLive will send a "Go Live" message to followers.

activitypub/workerpool/outbound.go

@@ -2,6 +2,8 @@ package workerpool
 
 import (
 	"net/http"
+	"sync"
+	"time"
 
 	log "github.com/sirupsen/logrus"
 )
@@ -13,10 +15,46 @@ type Job struct {
 
 var queue chan Job
 
+// Circuit breaker backoff durations for exponential backoff.
+var circuitBreakerBackoffDurations = []time.Duration{
+	1 * time.Minute,  // 1st failure: 1 minute
+	5 * time.Minute,  // 2nd failure: 5 minutes
+	15 * time.Minute, // 3rd failure: 15 minutes
+	30 * time.Minute, // 4th failure: 30 minutes
+	60 * time.Minute, // 5+ failures: 1 hour (max)
+}
+
+// httpClient is a configured HTTP client with timeouts and connection limits.
+var httpClient *http.Client
+
+// failedDomains tracks domains that are consistently failing with their failure count and backoff time.
+var (
+	failedDomains      = make(map[string]*domainFailure)
+	failedDomainsMutex sync.RWMutex
+)
+
+type domainFailure struct {
+	count        int
+	lastFailed   time.Time
+	backoffUntil time.Time
+}
+
 // InitOutboundWorkerPool starts n go routines that await ActivityPub jobs.
 func InitOutboundWorkerPool(workerPoolSize int) {
 	queue = make(chan Job, workerPoolSize)
 
+	// Initialize HTTP client with conservative timeouts and connection limits
+	// to prevent resource exhaustion and hanging requests
+	httpClient = &http.Client{
+		Timeout: 15 * time.Second, // Reduced from 30s for faster failure detection
+		Transport: &http.Transport{
+			MaxIdleConns:        20,               // Reduced from 100 to limit resource usage
+			MaxIdleConnsPerHost: 2,                // Reduced from 10 to be more conservative
+			IdleConnTimeout:     10 * time.Second, // Reduced from 30s for faster cleanup
+			DisableKeepAlives:   false,
+		},
+	}
+
 	// start workers
 	for i := 1; i <= workerPoolSize; i++ {
 		go worker(i, queue)
@@ -25,6 +63,12 @@ func InitOutboundWorkerPool(workerPoolSize int) {
 
 // AddToOutboundQueue will queue up an outbound http request.
 func AddToOutboundQueue(req *http.Request) {
+	// Check if domain should be skipped due to circuit breaker
+	if shouldSkipDomain(req.URL.Host) {
+		log.Debugf("Skipping request to %s due to circuit breaker", req.URL.Host)
+		return
+	}
+
 	select {
 	case queue <- Job{req}:
 	default:
@@ -40,20 +84,88 @@ func worker(workerID int, queue <-chan Job) {
 	for job := range queue {
 		if err := sendActivityPubMessageToInbox(job); err != nil {
 			log.Errorf("ActivityPub destination %s failed to send Error: %s", job.request.RequestURI, err)
+			recordDomainFailure(job.request.URL.Host)
+		} else {
+			// Reset domain failure count on success
+			resetDomainFailure(job.request.URL.Host)
 		}
 		log.Tracef("Done with ActivityPub destination %s using worker %d", job.request.RequestURI, workerID)
 	}
 }
 
 func sendActivityPubMessageToInbox(job Job) error {
-	client := &http.Client{}
+	resp, err := httpClient.Do(job.request)
-
-	resp, err := client.Do(job.request)
 	if err != nil {
 		return err
 	}
 
 	defer resp.Body.Close()
 
+	// Consider HTTP 4xx and 5xx as failures for circuit breaker purposes
+	if resp.StatusCode >= 400 {
+		return &httpError{statusCode: resp.StatusCode, message: resp.Status}
+	}
+
 	return nil
 }
+
+// httpError represents an HTTP error response.
+type httpError struct {
+	statusCode int
+	message    string
+}
+
+func (e *httpError) Error() string {
+	return e.message
+}
+
+// shouldSkipDomain checks if a domain should be skipped due to circuit breaker.
+func shouldSkipDomain(domain string) bool {
+	failedDomainsMutex.RLock()
+	defer failedDomainsMutex.RUnlock()
+
+	failure, exists := failedDomains[domain]
+	if !exists {
+		return false
+	}
+
+	// If we're still in backoff period, skip this domain
+	return time.Now().Before(failure.backoffUntil)
+}
+
+// recordDomainFailure records a failure for a domain and implements exponential backoff.
+func recordDomainFailure(domain string) {
+	failedDomainsMutex.Lock()
+	defer failedDomainsMutex.Unlock()
+
+	failure, exists := failedDomains[domain]
+	if !exists {
+		failure = &domainFailure{}
+		failedDomains[domain] = failure
+	}
+
+	failure.count++
+	failure.lastFailed = time.Now()
+
+	// Use exponential backoff with pre-defined durations
+	backoffIndex := failure.count - 1
+	if backoffIndex >= len(circuitBreakerBackoffDurations) {
+		backoffIndex = len(circuitBreakerBackoffDurations) - 1
+	}
+
+	backoffDuration := circuitBreakerBackoffDurations[backoffIndex]
+	failure.backoffUntil = time.Now().Add(backoffDuration)
+
+	log.Warnf("Domain %s failed %d times, backing off for %v", domain, failure.count, backoffDuration)
+}
+
+// resetDomainFailure resets the failure count for a domain on successful delivery.
+func resetDomainFailure(domain string) {
+	failedDomainsMutex.Lock()
+	defer failedDomainsMutex.Unlock()
+
+	if failure, exists := failedDomains[domain]; exists && failure.count > 0 {
+		log.Debugf("Resetting failure count for domain %s after successful delivery", domain)
+		delete(failedDomains, domain)
+	}
+}

activitypub/workerpool/outbound_test.go

@@ -0,0 +1,127 @@
+package workerpool
+
+import (
+	"testing"
+	"time"
+)
+
+// resetCircuitBreakerForTesting clears all failed domain state for testing purposes.
+// This function provides test isolation by resetting the global circuit breaker state.
+func resetCircuitBreakerForTesting() {
+	failedDomainsMutex.Lock()
+	defer failedDomainsMutex.Unlock()
+	failedDomains = make(map[string]*domainFailure)
+}
+
+func TestCircuitBreaker(t *testing.T) {
+	// Ensure clean state before test
+	resetCircuitBreakerForTesting()
+	defer resetCircuitBreakerForTesting() // Clean up after test
+
+	testDomain := "failing.example.com"
+
+	// Initially, domain should not be skipped
+	if shouldSkipDomain(testDomain) {
+		t.Error("Domain should not be skipped initially")
+	}
+
+	// Record failures
+	recordDomainFailure(testDomain)
+	recordDomainFailure(testDomain)
+	recordDomainFailure(testDomain)
+
+	// Domain should now be skipped
+	if !shouldSkipDomain(testDomain) {
+		t.Error("Domain should be skipped after failures")
+	}
+
+	// After successful delivery, domain should be reset
+	resetDomainFailure(testDomain)
+	if shouldSkipDomain(testDomain) {
+		t.Error("Domain should not be skipped after reset")
+	}
+}
+
+func TestHTTPTimeouts(t *testing.T) {
+	// Ensure clean state before test
+	resetCircuitBreakerForTesting()
+	defer resetCircuitBreakerForTesting() // Clean up after test
+
+	// Initialize HTTP client
+	InitOutboundWorkerPool(1)
+
+	if httpClient == nil {
+		t.Error("HTTP client should be initialized")
+	}
+
+	if httpClient.Timeout != 15*time.Second {
+		t.Error("HTTP client should have 15 second timeout")
+	}
+}
+
+func TestWorkerPoolSizing(t *testing.T) {
+	// Ensure clean state before test
+	resetCircuitBreakerForTesting()
+	defer resetCircuitBreakerForTesting() // Clean up after test
+
+	// Test the queue sizing
+	InitOutboundWorkerPool(5)
+
+	if cap(queue) != 5 {
+		t.Errorf("Queue capacity should be 5, got %d", cap(queue))
+	}
+}
+
+func TestBackoffDurations(t *testing.T) {
+	// Test that backoff durations are properly configured
+	expectedDurations := []time.Duration{
+		1 * time.Minute,
+		5 * time.Minute,
+		15 * time.Minute,
+		30 * time.Minute,
+		60 * time.Minute,
+	}
+
+	if len(circuitBreakerBackoffDurations) != len(expectedDurations) {
+		t.Errorf("Expected %d backoff durations, got %d", len(expectedDurations), len(circuitBreakerBackoffDurations))
+	}
+
+	for i, expected := range expectedDurations {
+		if circuitBreakerBackoffDurations[i] != expected {
+			t.Errorf("Backoff duration at index %d: expected %v, got %v", i, expected, circuitBreakerBackoffDurations[i])
+		}
+	}
+}
+
+func TestCircuitBreakerIsolation(t *testing.T) {
+	// Test that multiple tests don't interfere with each other
+	resetCircuitBreakerForTesting()
+	defer resetCircuitBreakerForTesting()
+
+	domain1 := "test1.example.com"
+	domain2 := "test2.example.com"
+
+	// Neither domain should be blocked initially
+	if shouldSkipDomain(domain1) || shouldSkipDomain(domain2) {
+		t.Error("Domains should not be blocked initially")
+	}
+
+	// Record failures for domain1 only
+	recordDomainFailure(domain1)
+	recordDomainFailure(domain1)
+	recordDomainFailure(domain1)
+
+	// Only domain1 should be blocked
+	if !shouldSkipDomain(domain1) {
+		t.Error("Domain1 should be blocked after failures")
+	}
+	if shouldSkipDomain(domain2) {
+		t.Error("Domain2 should not be blocked")
+	}
+
+	// Reset and verify clean state
+	resetCircuitBreakerForTesting()
+	if shouldSkipDomain(domain1) || shouldSkipDomain(domain2) {
+		t.Error("Both domains should be unblocked after reset")
+	}
+}