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/delete.go | 148 +++++++++++++++++++++++------ 1 file changed, 119 insertions(+), 29 deletions(-) (limited to 'internal/federation/federatingdb/delete.go') diff --git a/internal/federation/federatingdb/delete.go b/internal/federation/federatingdb/delete.go index 622ef6d3d..7e9b66c5a 100644 --- a/internal/federation/federatingdb/delete.go +++ b/internal/federation/federatingdb/delete.go @@ -19,10 +19,13 @@ package federatingdb import ( "context" + "errors" "net/url" - "codeberg.org/gruf/go-kv" "github.com/superseriousbusiness/gotosocial/internal/ap" + "github.com/superseriousbusiness/gotosocial/internal/db" + "github.com/superseriousbusiness/gotosocial/internal/gtserror" + "github.com/superseriousbusiness/gotosocial/internal/gtsmodel" "github.com/superseriousbusiness/gotosocial/internal/log" "github.com/superseriousbusiness/gotosocial/internal/messages" ) @@ -34,43 +37,130 @@ import ( // // The library makes this call only after acquiring a lock first. func (f *federatingDB) Delete(ctx context.Context, id *url.URL) error { - l := log.WithContext(ctx). - WithFields(kv.Fields{ - {"id", id}, - }...) - l.Debug("entering Delete") - activityContext := getActivityContext(ctx) if activityContext.internal { return nil // Already processed. } - requestingAcct := activityContext.requestingAcct - receivingAcct := activityContext.receivingAcct - - // in a delete we only get the URI, we can't know if we have a status or a profile or something else, - // so we have to try a few different things... - if s, err := f.state.DB.GetStatusByURI(ctx, id.String()); err == nil && requestingAcct.ID == s.AccountID { - l.Debugf("deleting status: %s", s.ID) - f.state.Workers.EnqueueFediAPI(ctx, messages.FromFediAPI{ - APObjectType: ap.ObjectNote, - APActivityType: ap.ActivityDelete, - GTSModel: s, - ReceivingAccount: receivingAcct, - RequestingAccount: requestingAcct, + // Extract receiving / requesting accounts. + requesting := activityContext.requestingAcct + receiving := activityContext.receivingAcct + + // Serialize deleted ID URI. + // (may be status OR account) + uriStr := id.String() + + var ( + ok bool + err error + ) + + // Try delete as an account URI. + ok, err = f.deleteAccount(ctx, + requesting, + receiving, + uriStr, + ) + if err != nil { + return err + } else if ok { + // success! + return nil + } + + // Try delete as a status URI. + ok, err = f.deleteStatus(ctx, + requesting, + receiving, + uriStr, + ) + if err != nil { + return err + } else if ok { + // success! + return nil + } + + // Log at debug level, as lots of these could indicate federation + // issues between remote and this instance, or help with debugging. + log.Debugf(ctx, "received delete for unknown target: %s", uriStr) + return nil +} + +func (f *federatingDB) deleteAccount( + ctx context.Context, + requesting *gtsmodel.Account, + receiving *gtsmodel.Account, + uri string, // target account +) ( + bool, // success? + error, // any error +) { + account, err := f.state.DB.GetAccountByURI(ctx, uri) + if err != nil && !errors.Is(err, db.ErrNoEntries) { + return false, gtserror.Newf("error getting account: %w", err) + } + + if account != nil { + // Ensure requesting account is + // only trying to delete itself. + if account.ID != requesting.ID { + + // TODO: handled forwarded deletes, + // for now we silently drop this. + return true, nil + } + + log.Debugf(ctx, "deleting account: %s", account.URI) + f.state.Workers.Federator.Queue.Push(&messages.FromFediAPI{ + APObjectType: ap.ObjectProfile, + APActivityType: ap.ActivityDelete, + GTSModel: account, + Receiving: receiving, + Requesting: requesting, }) + + return true, nil + } + + return false, nil +} + +func (f *federatingDB) deleteStatus( + ctx context.Context, + requesting *gtsmodel.Account, + receiving *gtsmodel.Account, + uri string, // target status +) ( + bool, // success? + error, // any error +) { + status, err := f.state.DB.GetStatusByURI(ctx, uri) + if err != nil && !errors.Is(err, db.ErrNoEntries) { + return false, gtserror.Newf("error getting status: %w", err) } - if a, err := f.state.DB.GetAccountByURI(ctx, id.String()); err == nil && requestingAcct.ID == a.ID { - l.Debugf("deleting account: %s", a.ID) - f.state.Workers.EnqueueFediAPI(ctx, messages.FromFediAPI{ - APObjectType: ap.ObjectProfile, - APActivityType: ap.ActivityDelete, - GTSModel: a, - ReceivingAccount: receivingAcct, - RequestingAccount: requestingAcct, + if status != nil { + // Ensure requesting account is only + // trying to delete its own statuses. + if status.AccountID != requesting.ID { + + // TODO: handled forwarded deletes, + // for now we silently drop this. + return true, nil + } + + log.Debugf(ctx, "deleting status: %s", status.URI) + f.state.Workers.Federator.Queue.Push(&messages.FromFediAPI{ + APObjectType: ap.ObjectNote, + APActivityType: ap.ActivityDelete, + GTSModel: status, + Receiving: receiving, + Requesting: requesting, }) + + return true, nil } - return nil + return false, nil } -- cgit v1.2.3