summaryrefslogtreecommitdiff
path: root/internal/processing/workers/federate.go
diff options
context:
space:
mode:
authorLibravatar kim <89579420+NyaaaWhatsUpDoc@users.noreply.github.com>2024-04-11 10:45:35 +0100
committerLibravatar GitHub <noreply@github.com>2024-04-11 11:45:35 +0200
commita483bd9e38333b153df1d4df95276cca38f99ff5 (patch)
treeb2fdb6f53ef248b31719a15adc93e767eba3d5c4 /internal/processing/workers/federate.go
parent[chore]: Bump github.com/yuin/goldmark from 1.7.0 to 1.7.1 (#2819) (diff)
downloadgotosocial-a483bd9e38333b153df1d4df95276cca38f99ff5.tar.xz
[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{}
Diffstat (limited to 'internal/processing/workers/federate.go')
-rw-r--r--internal/processing/workers/federate.go11
1 files changed, 11 insertions, 0 deletions
diff --git a/internal/processing/workers/federate.go b/internal/processing/workers/federate.go
index 9fdb8f662..e737513f5 100644
--- a/internal/processing/workers/federate.go
+++ b/internal/processing/workers/federate.go
@@ -75,6 +75,12 @@ func (f *federate) DeleteAccount(ctx context.Context, account *gtsmodel.Account)
return nil
}
+ // Drop any queued outgoing AP requests to / from account,
+ // (this stops any queued likes, boosts, creates etc).
+ f.state.Workers.Delivery.Queue.Delete("ActorID", account.URI)
+ f.state.Workers.Delivery.Queue.Delete("ObjectID", account.URI)
+ f.state.Workers.Delivery.Queue.Delete("TargetID", account.URI)
+
// Parse relevant URI(s).
outboxIRI, err := parseURI(account.OutboxURI)
if err != nil {
@@ -222,6 +228,11 @@ func (f *federate) DeleteStatus(ctx context.Context, status *gtsmodel.Status) er
return nil
}
+ // Drop any queued outgoing http requests for status,
+ // (this stops any queued likes, boosts, creates etc).
+ f.state.Workers.Delivery.Queue.Delete("ObjectID", status.URI)
+ f.state.Workers.Delivery.Queue.Delete("TargetID", status.URI)
+
// Ensure the status model is fully populated.
if err := f.state.DB.PopulateStatus(ctx, status); err != nil {
return gtserror.Newf("error populating status: %w", err)