From a483bd9e38333b153df1d4df95276cca38f99ff5 Mon Sep 17 00:00:00 2001
From: kim <89579420+NyaaaWhatsUpDoc@users.noreply.github.com>
Date: Thu, 11 Apr 2024 10:45:35 +0100
Subject: [performance] massively improved ActivityPub delivery worker
 efficiency (#2812)

* add delivery worker type that pulls from queue to httpclient package

* finish up some code commenting, bodge a vendored activity library change, integrate the deliverypool changes into transportcontroller

* hook up queue deletion logic

* support deleting queued http requests by target ID

* don't index APRequest by hostname in the queue

* use gorun

* use the original context's values when wrapping msg type as delivery{}

* actually log in the AP delivery worker ...

* add uncommitted changes

* use errors.AsV2()

* use errorsv2.AsV2()

* finish adding some code comments, add bad host handling to delivery workers

* slightly tweak deliveryworkerpool API, use advanced sender multiplier

* remove PopCtx() method, let others instead rely on Wait()

* shuffle things around to move delivery stuff into transport/ subpkg

* remove dead code

* formatting

* validate request before queueing for delivery

* finish adding code comments, fix up backoff code

* finish adding more code comments

* clamp minimum no. senders to 1

* add start/stop logging to delivery worker, some slight changes

* remove double logging

* use worker ptrs

* expose the embedded log fields in httpclient.Request{}

* ensure request context values are preserved when updating ctx

* add delivery worker tests

* fix linter issues

* ensure delivery worker gets inited in testrig

* fix tests to delivering messages to check worker delivery queue

* update error type to use ptr instead of value receiver

* fix test calling Workers{}.Start() instead of testrig.StartWorkers()

* update docs for advanced-sender-multiplier

* update to the latest activity library version

* add comment about not using httptest.Server{}
---
 testrig/util.go | 74 ++++++++++++++++++++++++++++++++++++++++++++++++++++-----
 1 file changed, 68 insertions(+), 6 deletions(-)

(limited to 'testrig/util.go')

diff --git a/testrig/util.go b/testrig/util.go
index e078074e5..f6f139e79 100644
--- a/testrig/util.go
+++ b/testrig/util.go
@@ -30,7 +30,7 @@ import (
 	"github.com/superseriousbusiness/gotosocial/internal/filter/visibility"
 	"github.com/superseriousbusiness/gotosocial/internal/messages"
 	tlprocessor "github.com/superseriousbusiness/gotosocial/internal/processing/timeline"
-	wprocessor "github.com/superseriousbusiness/gotosocial/internal/processing/workers"
+	"github.com/superseriousbusiness/gotosocial/internal/processing/workers"
 	"github.com/superseriousbusiness/gotosocial/internal/state"
 	"github.com/superseriousbusiness/gotosocial/internal/timeline"
 	"github.com/superseriousbusiness/gotosocial/internal/typeutils"
@@ -44,6 +44,8 @@ func StartNoopWorkers(state *state.State) {
 	state.Workers.ProcessFromClientAPI = func(context.Context, messages.FromClientAPI) error { return nil }
 	state.Workers.ProcessFromFediAPI = func(context.Context, messages.FromFediAPI) error { return nil }
 
+	state.Workers.Delivery.Init(nil)
+
 	_ = state.Workers.Scheduler.Start()
 	_ = state.Workers.ClientAPI.Start(1, 10)
 	_ = state.Workers.Federator.Start(1, 10)
@@ -52,11 +54,13 @@ func StartNoopWorkers(state *state.State) {
 
 // Starts workers on the provided state using processing functions from the given
 // workers processor. Useful when you *do* want to trigger side effects in a test.
-func StartWorkers(state *state.State, wProcessor *wprocessor.Processor) {
-	state.Workers.EnqueueClientAPI = wProcessor.EnqueueClientAPI
-	state.Workers.EnqueueFediAPI = wProcessor.EnqueueFediAPI
-	state.Workers.ProcessFromClientAPI = wProcessor.ProcessFromClientAPI
-	state.Workers.ProcessFromFediAPI = wProcessor.ProcessFromFediAPI
+func StartWorkers(state *state.State, processor *workers.Processor) {
+	state.Workers.EnqueueClientAPI = processor.EnqueueClientAPI
+	state.Workers.EnqueueFediAPI = processor.EnqueueFediAPI
+	state.Workers.ProcessFromClientAPI = processor.ProcessFromClientAPI
+	state.Workers.ProcessFromFediAPI = processor.ProcessFromFediAPI
+
+	state.Workers.Delivery.Init(nil)
 
 	_ = state.Workers.Scheduler.Start()
 	_ = state.Workers.ClientAPI.Start(1, 10)
@@ -93,6 +97,64 @@ func StartTimelines(state *state.State, filter *visibility.Filter, converter *ty
 	}
 }
 
+// EqualRequestURIs checks whether inputs have equal request URIs,
+// handling cases of url.URL{}, *url.URL{}, string, *string.
+func EqualRequestURIs(u1, u2 any) bool {
+	var uri1, uri2 string
+
+	requestURI := func(in string) (string, error) {
+		u, err := url.Parse(in)
+		if err != nil {
+			return "", err
+		}
+		return u.RequestURI(), nil
+	}
+
+	switch u1 := u1.(type) {
+	case url.URL:
+		uri1 = u1.RequestURI()
+	case *url.URL:
+		uri1 = u1.RequestURI()
+	case *string:
+		var err error
+		uri1, err = requestURI(*u1)
+		if err != nil {
+			return false
+		}
+	case string:
+		var err error
+		uri1, err = requestURI(u1)
+		if err != nil {
+			return false
+		}
+	default:
+		panic("unsupported type")
+	}
+
+	switch u2 := u2.(type) {
+	case url.URL:
+		uri2 = u2.RequestURI()
+	case *url.URL:
+		uri2 = u2.RequestURI()
+	case *string:
+		var err error
+		uri2, err = requestURI(*u2)
+		if err != nil {
+			return false
+		}
+	case string:
+		var err error
+		uri2, err = requestURI(u2)
+		if err != nil {
+			return false
+		}
+	default:
+		panic("unsupported type")
+	}
+
+	return uri1 == uri2
+}
+
 // CreateMultipartFormData is a handy function for taking a fieldname and a filename, and creating a multipart form bytes buffer
 // with the file contents set in the given fieldname. The extraFields param can be used to add extra FormFields to the request, as necessary.
 // The returned bytes.Buffer b can be used like so:
-- 
cgit v1.2.3