[performance] don't block on media load on API endpoints to improve latency (#4756)

Undoes my own recent change that blocks on media loading for timeline endpoints, to improve latency (which definitely had a bit of a regression after v0.21.0). Replaces the existing code that blocks on media loading in the surfacer with a more optimized check that reduces mutex locks and shouldn't lead to any accidental double loads.

Reviewed-on: https://codeberg.org/superseriousbusiness/gotosocial/pulls/4756
Co-authored-by: kim <grufwub@gmail.com>
Co-committed-by: kim <grufwub@gmail.com>

internal/federation/dereferencing/media.go

@@ -193,6 +193,86 @@ func (d *Dereferencer) RefreshMedia(
 	)
 }
 
+// WaitOnStatusMedia is a utility function to block until all status media have finished loading.
+// TODO: remove this temporary function with ingester{} work is underway, to instead stream status updates.
+func (d *Dereferencer) WaitOnStatusMedia(ctx context.Context, status *gtsmodel.Status) {
+	type uncachedMedia struct {
+		// Ptr to currently processing
+		// media to block on, if any.
+		Ptr *media.ProcessingMedia
+
+		// Database ID.
+		ID string
+
+		// Remote URL key.
+		URL string
+
+		// Index in status
+		// attachment slice.
+		Idx int
+	}
+
+	// Check if anything to be done.
+	if len(status.Attachments) == 0 {
+		return
+	}
+
+	// Append media in status attachments that isn't yet cached.
+	uncached := make([]uncachedMedia, 0, len(status.Attachments))
+	for i, media := range status.Attachments {
+		if !media.Cached() {
+			uncached = append(uncached, uncachedMedia{
+				ID:  media.ID,
+				URL: media.RemoteURL,
+				Idx: i,
+			})
+		}
+	}
+
+	// Check if any uncached.
+	if len(uncached) == 0 {
+		return
+	}
+
+	// To minimize mutex locks / unlocks,
+	// acquire all processing media at once.
+	d.derefMediaMu.Lock()
+
+	for i, entry := range uncached {
+		// Check for processing media by remote URL.
+		processing := d.derefMedia.get(entry.URL)
+		uncached[i].Ptr = processing
+	}
+
+	// Done with mutex lock.
+	d.derefMediaMu.Unlock()
+
+	for _, entry := range uncached {
+		if entry.Ptr != nil {
+			// If media was processing, block
+			// until finished loading. We don't
+			// care about error return as async
+			// thread will handle logging it,
+			// and media is always non-nil.
+			media, _ := entry.Ptr.Load(ctx)
+
+			// Set latest attachment on the status.
+			status.Attachments[entry.Idx] = media
+		} else {
+
+			// Media had finished processing, get latest from database.
+			media, err := d.state.DB.GetAttachmentByID(ctx, entry.ID)
+			if err != nil {
+				log.Errorf(ctx, "error getting latest attachment %s: %v", entry.URL, err)
+				continue
+			}
+
+			// Set latest attachment on the status.
+			status.Attachments[entry.Idx] = media
+		}
+	}
+}
+
 // processingMediaSafely provides concurrency-safe processing of
 // a media with given remote URL string. if a copy of the media is
 // not already being processed, the given 'process' callback will

internal/processing/common/account.go

@@ -146,9 +146,6 @@ func (p *Processor) GetAPIAccount(
 		// Only return sensitive account model _if_ requester = target.
 		apiAcc, err = p.converter.AccountToAPIAccountSensitive(ctx, target)
 	} else {
-		// Ensure account media attachments loaded.
-		p.LoadAccountMedia(ctx, requester, target)
-
 		// Else, fall back to returning the public account model.
 		apiAcc, err = p.converter.AccountToAPIAccountPublic(ctx, target)
 	}
@@ -258,9 +255,6 @@ func (p *Processor) getVisibleAPIAccounts(
 			continue
 		}
 
-		// Ensure account media attachments loaded.
-		p.LoadAccountMedia(ctx, requester, account)
-
 		// Convert the account to a public API model representation.
 		apiAcc, err := p.converter.AccountToAPIAccountPublic(ctx, account)
 		if err != nil {

internal/processing/common/media.go

@@ -22,7 +22,6 @@ import (
 	"errors"
 	"fmt"
 
-	"code.superseriousbusiness.org/gopkg/log"
 	"code.superseriousbusiness.org/gotosocial/internal/config"
 	"code.superseriousbusiness.org/gotosocial/internal/gtserror"
 	"code.superseriousbusiness.org/gotosocial/internal/gtsmodel"
@@ -111,115 +110,3 @@ func (p *Processor) StoreLocalEmoji(
 
 	return emoji, nil
 }
-
-// LoadAccountMedia ensures that all media for
-// given remote account is loaded where possible.
-func (p *Processor) LoadAccountMedia(
-	ctx context.Context,
-	requester *gtsmodel.Account,
-	account *gtsmodel.Account,
-) {
-	if requester == nil {
-		return
-	}
-	p.loadAccountMedia(ctx, requester, account)
-}
-
-// LoadStatusMedia ensures that all media for
-// given remote status is loaded where possible.
-func (p *Processor) LoadStatusMedia(
-	ctx context.Context,
-	requester *gtsmodel.Account,
-	status *gtsmodel.Status,
-) {
-	if requester == nil {
-		return
-	}
-	p.loadAccountMedia(ctx, requester, status.Account)
-	if status.BoostOfAccount != nil {
-		p.loadAccountMedia(ctx, requester, status.BoostOfAccount)
-	}
-	p.loadStatusMedia(ctx, requester, status)
-	if status.BoostOf != nil {
-		p.loadStatusMedia(ctx, requester, status.BoostOf)
-	}
-}
-
-// loadAccountMedia contains the
-// "meat" of LoadAccountMedia().
-func (p *Processor) loadAccountMedia(
-	ctx context.Context,
-	requester *gtsmodel.Account,
-	account *gtsmodel.Account,
-) {
-	if account.IsLocal() {
-		return
-	}
-
-	var err error
-
-	if account.HeaderMediaAttachment != nil {
-		// Ensure account header attachment is loaded and cached.
-		//
-		// If media attachment is still processing, this call will block.
-		account.HeaderMediaAttachment, err = p.federator.RefreshMedia(ctx,
-			requester.Username,
-			account.HeaderMediaAttachment,
-			media.AdditionalMediaInfo{},
-			false, // force
-			false, // async
-		)
-		if err != nil {
-			log.Errorf(ctx, "error refreshing header attachment %s: %v", account.HeaderMediaAttachment.RemoteURL, err)
-		}
-	}
-
-	if account.AvatarMediaAttachment != nil {
-		// Ensure account avatar attachment is loaded and cached.
-		//
-		// If media attachment is still processing, this call will block.
-		account.AvatarMediaAttachment, err = p.federator.RefreshMedia(ctx,
-			requester.Username,
-			account.AvatarMediaAttachment,
-			media.AdditionalMediaInfo{},
-			false, // force
-			false, // async
-		)
-		if err != nil {
-			log.Errorf(ctx, "error refreshing avatar attachment %s: %v", account.AvatarMediaAttachment.RemoteURL, err)
-		}
-	}
-}
-
-// loadStatusMedia contains the
-// "meat" of LoadStatusMedia().
-func (p *Processor) loadStatusMedia(
-	ctx context.Context,
-	requester *gtsmodel.Account,
-	status *gtsmodel.Status,
-) {
-	if !status.Flags.Local() {
-		return
-	}
-
-	// Ensure status media attachments are loaded,
-	// the below funcion checks if already cached.
-	//
-	// If media attachments are already processing
-	// from previous dereference, this will block.
-	for i, attach := range status.Attachments {
-		attach, err := p.federator.RefreshMedia(ctx,
-			requester.Username,
-			attach,
-			media.AdditionalMediaInfo{},
-			false, // force
-			false, // async
-		)
-		if err != nil {
-			log.Errorf(ctx, "error refreshing media attachment %s: %v", attach.RemoteURL, err)
-		}
-
-		// Set media attachment model.
-		status.Attachments[i] = attach
-	}
-}

internal/processing/common/status.go

@@ -271,9 +271,6 @@ func (p *Processor) GetAPIStatus(
 	apiStatus *apimodel.Status,
 	errWithCode gtserror.WithCode,
 ) {
-	// Ensure status media is cached locally.
-	p.LoadStatusMedia(ctx, requester, target)
-
 	// Convert the target status to frontend API model.
 	apiStatus, err := p.converter.StatusToAPIStatus(ctx,
 		target,
@@ -351,9 +348,6 @@ func (p *Processor) GetVisibleAPIStatuses(
 			continue
 		}
 
-		// Ensure status media is cached locally.
-		p.LoadStatusMedia(ctx, requester, status)
-
 		// Not muted or "hide" filtered. Convert to API status.
 		apiStatus, err := p.converter.StatusToAPIStatus(ctx,
 			status,

internal/processing/timeline/timeline.go

@@ -142,9 +142,6 @@ func (p *Processor) getStatusTimeline(
 				return nil, nil
 			}
 
-			// Ensure status media is cached locally.
-			p.c.LoadStatusMedia(ctx, requester, status)
-
 			// Finally, pass status to get converted to API model.
 			apiStatus, err := p.converter.StatusToAPIStatus(ctx,
 				status,

internal/surfacing/surfacetimeline.go

@@ -30,7 +30,6 @@ import (
 	"code.superseriousbusiness.org/gotosocial/internal/gtscontext"
 	"code.superseriousbusiness.org/gotosocial/internal/gtserror"
 	"code.superseriousbusiness.org/gotosocial/internal/gtsmodel"
-	"code.superseriousbusiness.org/gotosocial/internal/media"
 	"code.superseriousbusiness.org/gotosocial/internal/stream"
 	"code.superseriousbusiness.org/gotosocial/internal/util"
 )
@@ -647,8 +646,9 @@ func (s *Surfacer) prepareStatusForTimeline(
 		return nil, false, nil
 	}
 
-	// Ensure status media loaded.
-	s.loadStatusMedia(ctx, status)
+	// Ensure status media has finished loading, as caller
+	// to surfacer will have fetched from the dereferencer.
+	s.federator.Dereferencer.WaitOnStatusMedia(ctx, status)
 
 	// Attempt to convert status to frontend API model.
 	apiStatus, err = s.converter.StatusToAPIStatus(ctx,
@@ -772,76 +772,3 @@ func (s *Surfacer) RemoveRelationshipFromTimelines(ctx context.Context, timeline
 			RemoveByAccountIDs(targetAccountID)
 	}
 }
-
-// loadStatusMedia ensures that relevant account status media is loaded and cached locally.
-func (s *Surfacer) loadStatusMedia(ctx context.Context, status *gtsmodel.Status) {
-	account := status.Account
-	if account.IsLocal() {
-		return
-	}
-	s.loadAccountAttachments(ctx, status.Account)
-	if status.BoostOfAccount != nil {
-		s.loadAccountAttachments(ctx, status.BoostOfAccount)
-	}
-	s.loadStatusAttachments(ctx, status)
-	if status.BoostOf != nil {
-		s.loadStatusAttachments(ctx, status.BoostOf)
-	}
-}
-
-func (s *Surfacer) loadAccountAttachments(ctx context.Context, account *gtsmodel.Account) {
-	var err error
-	if account.HeaderMediaAttachment != nil {
-		// Ensure account header attachment is loaded and cached.
-		//
-		// If media attachment is still processing, this call will block.
-		account.HeaderMediaAttachment, err = s.federator.RefreshMedia(ctx,
-			"", // instance account
-			account.HeaderMediaAttachment,
-			media.AdditionalMediaInfo{},
-			false, // force
-			false, // async
-		)
-		if err != nil {
-			log.Errorf(ctx, "error refreshing boost header attachment %s: %v", account.HeaderMediaAttachment.RemoteURL, err)
-		}
-	}
-	if account.AvatarMediaAttachment != nil {
-		// Ensure account avatar attachment is loaded and cached.
-		//
-		// If media attachment is still processing, this call will block.
-		account.AvatarMediaAttachment, err = s.federator.RefreshMedia(ctx,
-			"", // instance account
-			account.AvatarMediaAttachment,
-			media.AdditionalMediaInfo{},
-			false, // force
-			false, // async
-		)
-		if err != nil {
-			log.Errorf(ctx, "error refreshing boost avatar attachment %s: %v", account.AvatarMediaAttachment.RemoteURL, err)
-		}
-	}
-}
-
-func (s *Surfacer) loadStatusAttachments(ctx context.Context, status *gtsmodel.Status) {
-	// Ensure status media attachments are loaded,
-	// the below funcion checks if already cached.
-	//
-	// If media attachments are already processing
-	// from previous dereference, this will block.
-	for i, attach := range status.Attachments {
-		attach, err := s.federator.RefreshMedia(ctx,
-			"", // as instance account
-			attach,
-			media.AdditionalMediaInfo{},
-			false, // force
-			false, // async
-		)
-		if err != nil {
-			log.Errorf(ctx, "error refreshing media attachment %s: %v", attach.RemoteURL, err)
-		}
-
-		// Set media attachment model.
-		status.Attachments[i] = attach
-	}
-}