From c9c0773f2c2363dcfa37e675b83ec3f0b49bd0d9 Mon Sep 17 00:00:00 2001 From: kim <89579420+NyaaaWhatsUpDoc@users.noreply.github.com> Date: Fri, 26 Apr 2024 13:50:46 +0100 Subject: [performance] update remaining worker pools to use queues (#2865) * start replacing client + federator + media workers with new worker + queue types * refactor federatingDB.Delete(), drop queued messages when deleting account / status * move all queue purging to the processor workers * undo toolchain updates * code comments, ensure dereferencer worker pool gets started * update gruf libraries in readme * start the job scheduler separately to the worker pools * reshuffle ordering or server.go + remove duplicate worker start / stop * update go-list version * fix vendoring * move queue invalidation to before wipeing / deletion, to ensure queued work not dropped * add logging to worker processing functions in testrig, don't start workers in unexpected places * update go-structr to add (+then rely on) QueueCtx{} type * ensure more worker pools get started properly in tests * fix remaining broken tests relying on worker queue logic * fix account test suite queue popping logic, ensure noop workers do not pull from queue * move back accidentally shuffled account deletion order * ensure error (non nil!!) gets passed in refactored federatingDB{}.Delete() * silently drop deletes from accounts not permitted to * don't warn log on forwarded deletes * make if else clauses easier to parse * use getFederatorMsg() * improved code comment * improved code comment re: requesting account delete checks * remove boolean result from worker start / stop since false = already running or already stopped * remove optional passed-in http.client * remove worker starting from the admin CLI commands (we don't need to handle side-effects) * update prune cli to start scheduler but not all of the workers * fix rebase issues * remove redundant return statements * i'm sorry sir linter --- internal/processing/admin/accountaction.go | 8 ++++---- internal/processing/admin/accountapprove.go | 6 +++--- internal/processing/admin/accountreject.go | 6 +++--- internal/processing/admin/actions.go | 9 ++++++--- internal/processing/admin/domainblock.go | 10 ++++------ internal/processing/admin/report.go | 6 +++--- 6 files changed, 23 insertions(+), 22 deletions(-) (limited to 'internal/processing/admin') diff --git a/internal/processing/admin/accountaction.go b/internal/processing/admin/accountaction.go index 155d8c0b4..7fd1047c4 100644 --- a/internal/processing/admin/accountaction.go +++ b/internal/processing/admin/accountaction.go @@ -80,13 +80,13 @@ func (p *Processor) accountActionSuspend( Text: text, }, func(ctx context.Context) gtserror.MultiError { - if err := p.state.Workers.ProcessFromClientAPI( + if err := p.state.Workers.Client.Process( ctx, - messages.FromClientAPI{ + &messages.FromClientAPI{ APObjectType: ap.ActorPerson, APActivityType: ap.ActivityDelete, - OriginAccount: adminAcct, - TargetAccount: targetAcct, + Origin: adminAcct, + Target: targetAcct, }, ); err != nil { errs := gtserror.NewMultiError(1) diff --git a/internal/processing/admin/accountapprove.go b/internal/processing/admin/accountapprove.go index c8a49e089..ebc91ee0c 100644 --- a/internal/processing/admin/accountapprove.go +++ b/internal/processing/admin/accountapprove.go @@ -54,12 +54,12 @@ func (p *Processor) AccountApprove( if !*user.Approved { // Process approval side effects asynschronously. - p.state.Workers.EnqueueClientAPI(ctx, messages.FromClientAPI{ + p.state.Workers.Client.Queue.Push(&messages.FromClientAPI{ APObjectType: ap.ActorPerson, APActivityType: ap.ActivityAccept, GTSModel: user, - OriginAccount: adminAcct, - TargetAccount: user.Account, + Origin: adminAcct, + Target: user.Account, }) } diff --git a/internal/processing/admin/accountreject.go b/internal/processing/admin/accountreject.go index eee2b2ff5..e7d54be41 100644 --- a/internal/processing/admin/accountreject.go +++ b/internal/processing/admin/accountreject.go @@ -101,12 +101,12 @@ func (p *Processor) AccountReject( } // Process rejection side effects asynschronously. - p.state.Workers.EnqueueClientAPI(ctx, messages.FromClientAPI{ + p.state.Workers.Client.Queue.Push(&messages.FromClientAPI{ APObjectType: ap.ActorPerson, APActivityType: ap.ActivityReject, GTSModel: deniedUser, - OriginAccount: adminAcct, - TargetAccount: user.Account, + Origin: adminAcct, + Target: user.Account, }) return apiAccount, nil diff --git a/internal/processing/admin/actions.go b/internal/processing/admin/actions.go index 0c959b1c5..968e45baa 100644 --- a/internal/processing/admin/actions.go +++ b/internal/processing/admin/actions.go @@ -23,6 +23,7 @@ import ( "sync" "time" + "github.com/superseriousbusiness/gotosocial/internal/gtscontext" "github.com/superseriousbusiness/gotosocial/internal/gtserror" "github.com/superseriousbusiness/gotosocial/internal/gtsmodel" "github.com/superseriousbusiness/gotosocial/internal/log" @@ -97,8 +98,10 @@ func (a *Actions) Run( // we're done modifying it for now. a.m.Unlock() - // Do the rest of the work asynchronously. - a.state.Workers.ClientAPI.Enqueue(func(ctx context.Context) { + go func() { + // Use a background context with existing values. + ctx = gtscontext.WithValues(context.Background(), ctx) + // Run the thing and collect errors. if errs := f(ctx); errs != nil { action.Errors = make([]string, 0, len(errs)) @@ -119,7 +122,7 @@ func (a *Actions) Run( if err := a.state.DB.UpdateAdminAction(ctx, action, "completed_at", "errors"); err != nil { log.Errorf(ctx, "db error marking action %s as completed: %q", actionKey, err) } - }) + }() return nil } diff --git a/internal/processing/admin/domainblock.go b/internal/processing/admin/domainblock.go index 9fe5dc847..2fe10c97b 100644 --- a/internal/processing/admin/domainblock.go +++ b/internal/processing/admin/domainblock.go @@ -201,15 +201,13 @@ func (p *Processor) domainBlockSideEffects( // process an account delete message to remove // that account's posts, media, etc. if err := p.rangeDomainAccounts(ctx, block.Domain, func(account *gtsmodel.Account) { - cMsg := messages.FromClientAPI{ + if err := p.state.Workers.Client.Process(ctx, &messages.FromClientAPI{ APObjectType: ap.ActorPerson, APActivityType: ap.ActivityDelete, GTSModel: block, - OriginAccount: account, - TargetAccount: account, - } - - if err := p.state.Workers.ProcessFromClientAPI(ctx, cMsg); err != nil { + Origin: account, + Target: account, + }); err != nil { errs.Append(err) } }); err != nil { diff --git a/internal/processing/admin/report.go b/internal/processing/admin/report.go index 32f05719b..d2186cfa2 100644 --- a/internal/processing/admin/report.go +++ b/internal/processing/admin/report.go @@ -140,12 +140,12 @@ func (p *Processor) ReportResolve(ctx context.Context, account *gtsmodel.Account } // Process side effects of closing the report. - p.state.Workers.EnqueueClientAPI(ctx, messages.FromClientAPI{ + p.state.Workers.Client.Queue.Push(&messages.FromClientAPI{ APObjectType: ap.ActivityFlag, APActivityType: ap.ActivityUpdate, GTSModel: report, - OriginAccount: account, - TargetAccount: report.Account, + Origin: account, + Target: report.Account, }) apimodelReport, err := p.converter.ReportToAdminAPIReport(ctx, updatedReport, account) -- cgit v1.2.3