A set of additional outbound ActivityPub outbound delivery optimizations (#4750)

* fix: remove optimistic UI updates in external actions to prevent race condition (#4711)

When adding/deleting external actions, the UI would sometimes show
duplicated or cloned rows due to a race condition between:
1. The optimistic setActions() call updating local state
2. The save() callback updating externalActions in context
3. The useEffect syncing actions from externalActions

This fix removes the optimistic updates and lets the server response
be the single source of truth, updating the UI only after the context
is updated via the save() success callback.

Fixes #4347

* fix(ap): additional outbound ActivityPub delivery optimizations

- check circuit breaker before signing any messages or any other work
- create outbound messages in batches and not all at once
- add small delay between outbound messages to spread it out a bit

* fix(ap): increase ap queue buffer size, reduce outbound http client timeout

* fix(ap): handle malformed inbox URIs

* fix(ap): batch outbox based on work performed not index

* feat(ap): add retryable activitypub outbound delivery client

---------

Co-authored-by: John Costa <jcosta@execonline.com>

activitypub/outbox/outbox.go

@@ -6,6 +6,7 @@ import (
 	"path/filepath"
 	"regexp"
 	"strings"
+	"time"
 
 	"github.com/go-fed/activity/streams"
 	"github.com/go-fed/activity/streams/vocab"
@@ -233,6 +234,7 @@ func getHashtagLinkHTMLFromTagString(baseHashtag string) string {
 }
 
 // SendToFollowers will send an arbitrary payload to all follower inboxes.
+// Requests are batched to prevent resource exhaustion when there are many followers.
 func SendToFollowers(payload []byte) error {
 	configRepository := configrepository.Get()
 	localActor := apmodels.MakeLocalIRIForAccount(configRepository.GetDefaultFederationUsername())
@@ -243,8 +245,28 @@ func SendToFollowers(payload []byte) error {
 		return errors.New("unable to fetch followers to send payload to")
 	}
 
-	for _, follower := range followers {
-		inbox, _ := url.Parse(follower.Inbox)
+	// Batch size and delay to prevent resource exhaustion during delivery.
+	// This spreads CPU load from cryptographic signing over time.
+	const batchSize = 50
+	const batchDelay = 100 * time.Millisecond
+
+	queued := 0
+	skipped := 0
+
+	for i, follower := range followers {
+		inbox, err := url.Parse(follower.Inbox)
+		if err != nil {
+			log.Errorln("unable to parse follower inbox URL", follower.Inbox, err)
+			continue
+		}
+
+		// Pre-check circuit breaker BEFORE expensive cryptographic signing.
+		// This saves CPU cycles for domains we know are failing.
+		if workerpool.ShouldSkipDomain(inbox.Host) {
+			skipped++
+			continue
+		}
+
 		req, err := crypto.CreateSignedRequest(payload, inbox, localActor)
 		if err != nil {
 			log.Errorln("unable to create outbox request", follower.Inbox, err)
@@ -252,7 +274,21 @@ func SendToFollowers(payload []byte) error {
 		}
 
 		workerpool.AddToOutboundQueue(req)
+		queued++
+
+		// Add a small delay between batches to spread out CPU and network load.
+		// This helps prevent ActivityPub delivery from competing with video encoding.
+		// Use queued count (not loop index) to ensure consistent rate limiting
+		// even when followers are skipped due to circuit breaker or parse errors.
+		if queued%batchSize == 0 && i+1 < len(followers) {
+			time.Sleep(batchDelay)
+		}
 	}
+
+	if skipped > 0 {
+		log.Debugf("Skipped %d followers due to circuit breaker, queued %d", skipped, queued)
+	}
+
 	return nil
 }
 

activitypub/resolvers/resolve.go

@@ -11,6 +11,7 @@ import (
 	"github.com/owncast/owncast/activitypub/apmodels"
 	"github.com/owncast/owncast/activitypub/crypto"
 	"github.com/owncast/owncast/persistence/configrepository"
+	"github.com/owncast/owncast/utils"
 	"github.com/pkg/errors"
 	log "github.com/sirupsen/logrus"
 )
@@ -45,31 +46,37 @@ func Resolve(c context.Context, data []byte, callbacks ...interface{}) error {
 	return nil
 }
 
-// ResolveIRI will resolve an IRI ahd call the correct callback for the resolved type.
+// ResolveIRI will resolve an IRI and call the correct callback for the resolved type.
+// Uses a retryable HTTP client for resilience against transient failures.
 func ResolveIRI(c context.Context, iri string, callbacks ...interface{}) error {
 	configRepository := configrepository.Get()
 	log.Debugln("Resolving", iri)
 
 	req, _ := http.NewRequest(http.MethodGet, iri, nil)
+	req.Header.Set("Accept", "application/activity+json, application/ld+json")
 
 	actor := apmodels.MakeLocalIRIForAccount(configRepository.GetDefaultFederationUsername())
 	if err := crypto.SignRequest(req, nil, actor); err != nil {
 		return err
 	}
 
-	response, err := http.DefaultClient.Do(req)
+	client := utils.GetRetryableHTTPClient()
+	response, err := client.Do(req)
 	if err != nil {
 		return err
 	}
 
 	defer response.Body.Close()
 
+	if response.StatusCode >= 400 {
+		return errors.New("request failed with status: " + response.Status)
+	}
+
 	data, err := io.ReadAll(response.Body)
 	if err != nil {
 		return err
 	}
 
-	// fmt.Println(string(data))
 	return Resolve(c, data, callbacks...)
 }
 

activitypub/workerpool/outbound.go

@@ -5,6 +5,7 @@ import (
 	"sync"
 	"time"
 
+	"github.com/owncast/owncast/utils"
 	log "github.com/sirupsen/logrus"
 )
 
@@ -41,19 +42,18 @@ type domainFailure struct {
 
 // 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,
-		},
+	// Use a larger buffer to decouple request creation from processing.
+	// This prevents SendToFollowers from blocking when many followers need updates.
+	const minQueueBuffer = 500
+	queueBuffer := workerPoolSize * 10
+	if queueBuffer < minQueueBuffer {
+		queueBuffer = minQueueBuffer
 	}
+	queue = make(chan Job, queueBuffer)
+
+	// Initialize HTTP client with retry logic for transient failures
+	// The retryable client handles 502/503/504 errors automatically
+	httpClient = utils.GetRetryableHTTPClient()
 
 	// start workers
 	for i := 1; i <= workerPoolSize; i++ {
@@ -64,7 +64,7 @@ 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) {
+	if ShouldSkipDomain(req.URL.Host) {
 		log.Debugf("Skipping request to %s due to circuit breaker", req.URL.Host)
 		return
 	}
@@ -119,8 +119,9 @@ 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 {
+// ShouldSkipDomain checks if a domain should be skipped due to circuit breaker.
+// This is exported so callers can check before expensive operations like request signing.
+func ShouldSkipDomain(domain string) bool {
 	failedDomainsMutex.RLock()
 	defer failedDomainsMutex.RUnlock()
 

activitypub/workerpool/outbound_test.go

@@ -21,7 +21,7 @@ func TestCircuitBreaker(t *testing.T) {
 	testDomain := "failing.example.com"
 
 	// Initially, domain should not be skipped
-	if shouldSkipDomain(testDomain) {
+	if ShouldSkipDomain(testDomain) {
 		t.Error("Domain should not be skipped initially")
 	}
 
@@ -31,13 +31,13 @@ func TestCircuitBreaker(t *testing.T) {
 	recordDomainFailure(testDomain)
 
 	// Domain should now be skipped
-	if !shouldSkipDomain(testDomain) {
+	if !ShouldSkipDomain(testDomain) {
 		t.Error("Domain should be skipped after failures")
 	}
 
 	// After successful delivery, domain should be reset
 	resetDomainFailure(testDomain)
-	if shouldSkipDomain(testDomain) {
+	if ShouldSkipDomain(testDomain) {
 		t.Error("Domain should not be skipped after reset")
 	}
 }
@@ -54,8 +54,8 @@ func TestHTTPTimeouts(t *testing.T) {
 		t.Error("HTTP client should be initialized")
 	}
 
-	if httpClient.Timeout != 15*time.Second {
-		t.Error("HTTP client should have 15 second timeout")
+	if httpClient.Timeout != 8*time.Second {
+		t.Errorf("HTTP client should have 8 second timeout, got %v", httpClient.Timeout)
 	}
 }
 
@@ -64,11 +64,17 @@ func TestWorkerPoolSizing(t *testing.T) {
 	resetCircuitBreakerForTesting()
 	defer resetCircuitBreakerForTesting() // Clean up after test
 
-	// Test the queue sizing
+	// Test that queue buffer is at least the minimum (500) even for small worker pools
 	InitOutboundWorkerPool(5)
 
-	if cap(queue) != 5 {
-		t.Errorf("Queue capacity should be 5, got %d", cap(queue))
+	if cap(queue) < 500 {
+		t.Errorf("Queue capacity should be at least 500, got %d", cap(queue))
+	}
+
+	// Test that larger worker pools get proportionally larger buffers
+	InitOutboundWorkerPool(100)
+	if cap(queue) != 1000 {
+		t.Errorf("Queue capacity should be 1000 for 100 workers, got %d", cap(queue))
 	}
 }
 
@@ -102,7 +108,7 @@ func TestCircuitBreakerIsolation(t *testing.T) {
 	domain2 := "test2.example.com"
 
 	// Neither domain should be blocked initially
-	if shouldSkipDomain(domain1) || shouldSkipDomain(domain2) {
+	if ShouldSkipDomain(domain1) || ShouldSkipDomain(domain2) {
 		t.Error("Domains should not be blocked initially")
 	}
 
@@ -112,16 +118,16 @@ func TestCircuitBreakerIsolation(t *testing.T) {
 	recordDomainFailure(domain1)
 
 	// Only domain1 should be blocked
-	if !shouldSkipDomain(domain1) {
+	if !ShouldSkipDomain(domain1) {
 		t.Error("Domain1 should be blocked after failures")
 	}
-	if shouldSkipDomain(domain2) {
+	if ShouldSkipDomain(domain2) {
 		t.Error("Domain2 should not be blocked")
 	}
 
 	// Reset and verify clean state
 	resetCircuitBreakerForTesting()
-	if shouldSkipDomain(domain1) || shouldSkipDomain(domain2) {
+	if ShouldSkipDomain(domain1) || ShouldSkipDomain(domain2) {
 		t.Error("Both domains should be unblocked after reset")
 	}
 }

go.mod

@@ -54,6 +54,8 @@ require (
 	github.com/golang-jwt/jwt/v5 v5.2.1 // indirect
 	github.com/google/uuid v1.5.0 // indirect
 	github.com/gorilla/css v1.0.1 // indirect
+	github.com/hashicorp/go-cleanhttp v0.5.2 // indirect
+	github.com/hashicorp/go-retryablehttp v0.7.8 // indirect
 	github.com/jmespath/go-jmespath v0.4.0 // indirect
 	github.com/jonboulle/clockwork v0.5.0 // indirect
 	github.com/klauspost/compress v1.18.0 // indirect

go.sum

@@ -52,6 +52,10 @@ github.com/gorilla/websocket v1.5.3 h1:saDtZ6Pbx/0u+bgYQ3q96pZgCzfhKXGPqt7kZ72aN
 github.com/gorilla/websocket v1.5.3/go.mod h1:YR8l580nyteQvAITg2hZ9XVh4b55+EU/adAjf1fMHhE=
 github.com/grafov/m3u8 v0.12.1 h1:DuP1uA1kvRRmGNAZ0m+ObLv1dvrfNO0TPx0c/enNk0s=
 github.com/grafov/m3u8 v0.12.1/go.mod h1:nqzOkfBiZJENr52zTVd/Dcl03yzphIMbJqkXGu+u080=
+github.com/hashicorp/go-cleanhttp v0.5.2 h1:035FKYIWjmULyFRBKPs8TBQoi0x6d9G4xc9neXJWAZQ=
+github.com/hashicorp/go-cleanhttp v0.5.2/go.mod h1:kO/YDlP8L1346E6Sodw+PrpBSV4/SoxCXGY6BqNFT48=
+github.com/hashicorp/go-retryablehttp v0.7.8 h1:ylXZWnqa7Lhqpk0L1P1LzDtGcCR0rPVUrx/c8Unxc48=
+github.com/hashicorp/go-retryablehttp v0.7.8/go.mod h1:rjiScheydd+CxvumBsIrFKlx3iS0jrZ7LvzFGFmuKbw=
 github.com/jellydator/ttlcache/v3 v3.4.0 h1:YS4P125qQS0tNhtL6aeYkheEaB/m8HCqdMMP4mnWdTY=
 github.com/jellydator/ttlcache/v3 v3.4.0/go.mod h1:Hw9EgjymziQD3yGsQdf1FqFdpp7YjFMd4Srg5EJlgD4=
 github.com/jmespath/go-jmespath v0.4.0 h1:BEgLn5cpjn8UN1mAw4NjwDrS35OdebyEtFe+9YPoQUg=

utils/tlsconfig.go

@@ -0,0 +1,73 @@
+package utils
+
+import (
+	"crypto/tls"
+	"net/http"
+	"os"
+	"sync"
+	"time"
+
+	"github.com/hashicorp/go-retryablehttp"
+	log "github.com/sirupsen/logrus"
+)
+
+var (
+	insecureSkipVerify     bool
+	insecureSkipVerifyOnce sync.Once
+)
+
+// IsInsecureSkipVerifyEnabled returns true if the OWNCAST_INSECURE_SKIP_VERIFY
+// environment variable is set to "true". This is intended for testing only.
+func IsInsecureSkipVerifyEnabled() bool {
+	insecureSkipVerifyOnce.Do(func() {
+		insecureSkipVerify = os.Getenv("OWNCAST_INSECURE_SKIP_VERIFY") == "true"
+		if insecureSkipVerify {
+			log.Warnln("OWNCAST_INSECURE_SKIP_VERIFY is enabled - TLS certificate verification disabled (testing only)")
+		}
+	})
+	return insecureSkipVerify
+}
+
+// GetTLSConfig returns a TLS config that optionally skips certificate verification
+// based on the OWNCAST_INSECURE_SKIP_VERIFY environment variable.
+func GetTLSConfig() *tls.Config {
+	if IsInsecureSkipVerifyEnabled() {
+		return &tls.Config{
+			InsecureSkipVerify: true, // #nosec G402 - intentional for testing
+		}
+	}
+	return nil
+}
+
+// GetHTTPTransportWithTLS returns an http.Transport configured with TLS settings.
+// If OWNCAST_INSECURE_SKIP_VERIFY is set, certificate verification is skipped.
+func GetHTTPTransportWithTLS(baseTransport *http.Transport) *http.Transport {
+	if baseTransport == nil {
+		baseTransport = &http.Transport{}
+	}
+	baseTransport.TLSClientConfig = GetTLSConfig()
+	return baseTransport
+}
+
+// GetRetryableHTTPClient returns an http.Client with retry logic for transient failures.
+// It uses hashicorp/go-retryablehttp with exponential backoff for 502, 503, 504 errors.
+func GetRetryableHTTPClient() *http.Client {
+	retryClient := retryablehttp.NewClient()
+	retryClient.RetryMax = 3
+	retryClient.RetryWaitMin = 100 * time.Millisecond
+	retryClient.RetryWaitMax = 1 * time.Second
+	retryClient.Logger = nil // Disable default logging
+
+	// Configure transport with connection pooling limits
+	transport := GetHTTPTransportWithTLS(&http.Transport{
+		MaxIdleConns:        20,               // Limit resource usage
+		MaxIdleConnsPerHost: 2,                // Conservative per-host limit
+		IdleConnTimeout:     10 * time.Second, // Fast cleanup of idle connections
+		DisableKeepAlives:   false,
+	})
+	retryClient.HTTPClient.Transport = transport
+
+	client := retryClient.StandardClient()
+	client.Timeout = 8 * time.Second // Short timeout - legitimate servers respond quickly
+	return client
+}