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/federation/federatingdb/create.go | 92 +++++++++++++++--------------- 1 file changed, 46 insertions(+), 46 deletions(-) (limited to 'internal/federation/federatingdb/create.go') diff --git a/internal/federation/federatingdb/create.go b/internal/federation/federatingdb/create.go index 94261526e..44f3cd98c 100644 --- a/internal/federation/federatingdb/create.go +++ b/internal/federation/federatingdb/create.go @@ -99,7 +99,7 @@ func (f *federatingDB) Create(ctx context.Context, asType vocab.Type) error { BLOCK HANDLERS */ -func (f *federatingDB) activityBlock(ctx context.Context, asType vocab.Type, receiving *gtsmodel.Account, requestingAccount *gtsmodel.Account) error { +func (f *federatingDB) activityBlock(ctx context.Context, asType vocab.Type, receiving *gtsmodel.Account, requesting *gtsmodel.Account) error { blockable, ok := asType.(vocab.ActivityStreamsBlock) if !ok { return errors.New("activityBlock: could not convert type to block") @@ -110,10 +110,10 @@ func (f *federatingDB) activityBlock(ctx context.Context, asType vocab.Type, rec return fmt.Errorf("activityBlock: could not convert Block to gts model block") } - if block.AccountID != requestingAccount.ID { + if block.AccountID != requesting.ID { return fmt.Errorf( "activityBlock: requestingAccount %s is not Block actor account %s", - requestingAccount.URI, block.Account.URI, + requesting.URI, block.Account.URI, ) } @@ -130,12 +130,12 @@ func (f *federatingDB) activityBlock(ctx context.Context, asType vocab.Type, rec return fmt.Errorf("activityBlock: database error inserting block: %s", err) } - f.state.Workers.EnqueueFediAPI(ctx, messages.FromFediAPI{ - APObjectType: ap.ActivityBlock, - APActivityType: ap.ActivityCreate, - GTSModel: block, - ReceivingAccount: receiving, - RequestingAccount: requestingAccount, + f.state.Workers.Federator.Queue.Push(&messages.FromFediAPI{ + APObjectType: ap.ActivityBlock, + APActivityType: ap.ActivityCreate, + GTSModel: block, + Receiving: receiving, + Requesting: requesting, }) return nil @@ -297,7 +297,7 @@ func (f *federatingDB) createPollOptionables( } // Enqueue message to the fedi API worker with poll vote(s). - f.state.Workers.EnqueueFediAPI(ctx, messages.FromFediAPI{ + f.state.Workers.Federator.Queue.Push(&messages.FromFediAPI{ APActivityType: ap.ActivityCreate, APObjectType: ap.ActivityQuestion, GTSModel: >smodel.PollVote{ @@ -308,8 +308,8 @@ func (f *federatingDB) createPollOptionables( PollID: inReplyTo.PollID, Poll: inReplyTo.Poll, }, - ReceivingAccount: receiver, - RequestingAccount: requester, + Receiving: receiver, + Requesting: requester, }) return nil @@ -377,28 +377,28 @@ func (f *federatingDB) createStatusable( // Pass the statusable URI (APIri) into the processor // worker and do the rest of the processing asynchronously. - f.state.Workers.EnqueueFediAPI(ctx, messages.FromFediAPI{ - APObjectType: ap.ObjectNote, - APActivityType: ap.ActivityCreate, - APIri: ap.GetJSONLDId(statusable), - APObjectModel: nil, - GTSModel: nil, - ReceivingAccount: receiver, - RequestingAccount: requester, + f.state.Workers.Federator.Queue.Push(&messages.FromFediAPI{ + APObjectType: ap.ObjectNote, + APActivityType: ap.ActivityCreate, + APIRI: ap.GetJSONLDId(statusable), + APObject: nil, + GTSModel: nil, + Receiving: receiver, + Requesting: requester, }) return nil } // Do the rest of the processing asynchronously. The processor // will handle inserting/updating + further dereferencing the status. - f.state.Workers.EnqueueFediAPI(ctx, messages.FromFediAPI{ - APObjectType: ap.ObjectNote, - APActivityType: ap.ActivityCreate, - APIri: nil, - GTSModel: nil, - APObjectModel: statusable, - ReceivingAccount: receiver, - RequestingAccount: requester, + f.state.Workers.Federator.Queue.Push(&messages.FromFediAPI{ + APObjectType: ap.ObjectNote, + APActivityType: ap.ActivityCreate, + APIRI: nil, + GTSModel: nil, + APObject: statusable, + Receiving: receiver, + Requesting: requester, }) return nil @@ -439,12 +439,12 @@ func (f *federatingDB) activityFollow(ctx context.Context, asType vocab.Type, re return fmt.Errorf("activityFollow: database error inserting follow request: %s", err) } - f.state.Workers.EnqueueFediAPI(ctx, messages.FromFediAPI{ - APObjectType: ap.ActivityFollow, - APActivityType: ap.ActivityCreate, - GTSModel: followRequest, - ReceivingAccount: receivingAccount, - RequestingAccount: requestingAccount, + f.state.Workers.Federator.Queue.Push(&messages.FromFediAPI{ + APObjectType: ap.ActivityFollow, + APActivityType: ap.ActivityCreate, + GTSModel: followRequest, + Receiving: receivingAccount, + Requesting: requestingAccount, }) return nil @@ -484,12 +484,12 @@ func (f *federatingDB) activityLike(ctx context.Context, asType vocab.Type, rece return fmt.Errorf("activityLike: database error inserting fave: %w", err) } - f.state.Workers.EnqueueFediAPI(ctx, messages.FromFediAPI{ - APObjectType: ap.ActivityLike, - APActivityType: ap.ActivityCreate, - GTSModel: fave, - ReceivingAccount: receivingAccount, - RequestingAccount: requestingAccount, + f.state.Workers.Federator.Queue.Push(&messages.FromFediAPI{ + APObjectType: ap.ActivityLike, + APActivityType: ap.ActivityCreate, + GTSModel: fave, + Receiving: receivingAccount, + Requesting: requestingAccount, }) return nil @@ -536,12 +536,12 @@ func (f *federatingDB) activityFlag(ctx context.Context, asType vocab.Type, rece return fmt.Errorf("activityFlag: database error inserting report: %w", err) } - f.state.Workers.EnqueueFediAPI(ctx, messages.FromFediAPI{ - APObjectType: ap.ActivityFlag, - APActivityType: ap.ActivityCreate, - GTSModel: report, - ReceivingAccount: receivingAccount, - RequestingAccount: requestingAccount, + f.state.Workers.Federator.Queue.Push(&messages.FromFediAPI{ + APObjectType: ap.ActivityFlag, + APActivityType: ap.ActivityCreate, + GTSModel: report, + Receiving: receivingAccount, + Requesting: requestingAccount, }) return nil -- cgit v1.2.3