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{} --- internal/processing/followrequest_test.go | 83 +++++++++++++++++-------------- 1 file changed, 47 insertions(+), 36 deletions(-) (limited to 'internal/processing/followrequest_test.go') diff --git a/internal/processing/followrequest_test.go b/internal/processing/followrequest_test.go index 4c089be4a..db0419522 100644 --- a/internal/processing/followrequest_test.go +++ b/internal/processing/followrequest_test.go @@ -21,6 +21,7 @@ import ( "context" "encoding/json" "fmt" + "io" "testing" "time" @@ -77,22 +78,6 @@ func (suite *FollowRequestTestSuite) TestFollowRequestAccept() { Note: "", }, relationship) - // accept should be sent to Some_User - var sent [][]byte - if !testrig.WaitFor(func() bool { - sentI, ok := suite.httpClient.SentMessages.Load(targetAccount.InboxURI) - if ok { - sent, ok = sentI.([][]byte) - if !ok { - panic("SentMessages entry was not []byte") - } - return true - } - return false - }) { - suite.FailNow("timed out waiting for message") - } - accept := &struct { Actor string `json:"actor"` ID string `json:"id"` @@ -106,8 +91,29 @@ func (suite *FollowRequestTestSuite) TestFollowRequestAccept() { To string `json:"to"` Type string `json:"type"` }{} - err = json.Unmarshal(sent[0], accept) - suite.NoError(err) + + // accept should be sent to Some_User + var sent []byte + if !testrig.WaitFor(func() bool { + delivery, ok := suite.state.Workers.Delivery.Queue.Pop() + if !ok { + return false + } + if !testrig.EqualRequestURIs(delivery.Request.URL, targetAccount.InboxURI) { + panic("differing request uris") + } + sent, err = io.ReadAll(delivery.Request.Body) + if err != nil { + panic("error reading body: " + err.Error()) + } + err = json.Unmarshal(sent, accept) + if err != nil { + panic("error unmarshaling json: " + err.Error()) + } + return true + }) { + suite.FailNow("timed out waiting for message") + } suite.Equal(requestingAccount.URI, accept.Actor) suite.Equal(targetAccount.URI, accept.Object.Actor) @@ -144,22 +150,6 @@ func (suite *FollowRequestTestSuite) TestFollowRequestReject() { suite.NoError(errWithCode) suite.EqualValues(&apimodel.Relationship{ID: "01FHMQX3GAABWSM0S2VZEC2SWC", Following: false, ShowingReblogs: false, Notifying: false, FollowedBy: false, Blocking: false, BlockedBy: false, Muting: false, MutingNotifications: false, Requested: false, DomainBlocking: false, Endorsed: false, Note: ""}, relationship) - // reject should be sent to Some_User - var sent [][]byte - if !testrig.WaitFor(func() bool { - sentI, ok := suite.httpClient.SentMessages.Load(targetAccount.InboxURI) - if ok { - sent, ok = sentI.([][]byte) - if !ok { - panic("SentMessages entry was not []byte") - } - return true - } - return false - }) { - suite.FailNow("timed out waiting for message") - } - reject := &struct { Actor string `json:"actor"` ID string `json:"id"` @@ -173,8 +163,29 @@ func (suite *FollowRequestTestSuite) TestFollowRequestReject() { To string `json:"to"` Type string `json:"type"` }{} - err = json.Unmarshal(sent[0], reject) - suite.NoError(err) + + // reject should be sent to Some_User + var sent []byte + if !testrig.WaitFor(func() bool { + delivery, ok := suite.state.Workers.Delivery.Queue.Pop() + if !ok { + return false + } + if !testrig.EqualRequestURIs(delivery.Request.URL, targetAccount.InboxURI) { + panic("differing request uris") + } + sent, err = io.ReadAll(delivery.Request.Body) + if err != nil { + panic("error reading body: " + err.Error()) + } + err = json.Unmarshal(sent, reject) + if err != nil { + panic("error unmarshaling json: " + err.Error()) + } + return true + }) { + suite.FailNow("timed out waiting for message") + } suite.Equal(requestingAccount.URI, reject.Actor) suite.Equal(targetAccount.URI, reject.Object.Actor) -- cgit v1.2.3