From 5f81198328ad9d46f06691032706631ab054d657 Mon Sep 17 00:00:00 2001 From: kim Date: Wed, 24 Apr 2024 12:54:51 +0100 Subject: [PATCH] move all queue purging to the processor workers --- internal/federation/federatingdb/delete.go | 34 -------------- internal/processing/workers/federate.go | 35 --------------- internal/processing/workers/fromclientapi.go | 47 ++++++++++++++++++-- internal/processing/workers/fromfediapi.go | 46 +++++++++++++++++++ 4 files changed, 90 insertions(+), 72 deletions(-) diff --git a/internal/federation/federatingdb/delete.go b/internal/federation/federatingdb/delete.go index 8bfaa325a..2b06e2933 100644 --- a/internal/federation/federatingdb/delete.go +++ b/internal/federation/federatingdb/delete.go @@ -101,24 +101,6 @@ func (f *federatingDB) deleteAccount( } log.Debugf(ctx, "deleting account: %s", account.URI) - - // Drop any outgoing queued AP requests to / from / targeting - // this account, (stops queued likes, boosts, creates etc). - f.state.Workers.Delivery.Queue.Delete("ObjectID", account.URI) - f.state.Workers.Delivery.Queue.Delete("TargetID", account.URI) - - // Drop any incoming queued client messages to / from this - // account, (stops processing of local origin data for acccount). - f.state.Workers.Client.Queue.Delete("Target.ID", account.ID) - f.state.Workers.Client.Queue.Delete("TargetURI", account.URI) - - // Drop any incoming queued federator messages to this account, - // (stops processing of remote origin data targeting this account). - f.state.Workers.Federator.Queue.Delete("Requesting.ID", account.ID) - f.state.Workers.Federator.Queue.Delete("TargetURI", account.URI) - - // Only AFTER we have finished purging queues do we enqueue, - // otherwise we risk purging our own delete message from queue! f.state.Workers.Federator.Queue.Push(&messages.FromFediAPI{ APObjectType: ap.ObjectProfile, APActivityType: ap.ActivityDelete, @@ -154,22 +136,6 @@ func (f *federatingDB) deleteStatus( } log.Debugf(ctx, "deleting status: %s", status.URI) - - // Drop any outgoing queued AP requests about / targeting - // this status, (stops queued likes, boosts, creates etc). - f.state.Workers.Delivery.Queue.Delete("ObjectID", status.URI) - f.state.Workers.Delivery.Queue.Delete("TargetID", status.URI) - - // Drop any incoming queued client messages about / targeting - // status, (stops processing of local origin data for status). - f.state.Workers.Client.Queue.Delete("TargetURI", status.URI) - - // Drop any incoming queued federator messages targeting status, - // (stops processing of remote origin data targeting this status). - f.state.Workers.Federator.Queue.Delete("TargetURI", status.URI) - - // Only AFTER we have finished purging queues do we enqueue, - // otherwise we risk purging our own delete message from queue! f.state.Workers.Federator.Queue.Push(&messages.FromFediAPI{ APObjectType: ap.ObjectNote, APActivityType: ap.ActivityDelete, diff --git a/internal/processing/workers/federate.go b/internal/processing/workers/federate.go index d560ee1e3..3538c9958 100644 --- a/internal/processing/workers/federate.go +++ b/internal/processing/workers/federate.go @@ -96,23 +96,6 @@ func (f *federate) DeleteAccount(ctx context.Context, account *gtsmodel.Account) return err } - // Drop any outgoing queued AP requests to / from / targeting - // this account, (stops 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) - - // Drop any incoming queued client messages to / from this - // account, (stops processing of local origin data for acccount). - f.state.Workers.Client.Queue.Delete("Origin.ID", account.ID) - f.state.Workers.Client.Queue.Delete("Target.ID", account.ID) - f.state.Workers.Client.Queue.Delete("TargetURI", account.URI) - - // Drop any incoming queued federator messages to this account, - // (stops processing of remote origin data targeting this account). - f.state.Workers.Federator.Queue.Delete("Receiving.ID", account.ID) - f.state.Workers.Federator.Queue.Delete("TargetURI", account.URI) - // Create a new delete. // todo: tc.AccountToASDelete delete := streams.NewActivityStreamsDelete() @@ -245,24 +228,6 @@ func (f *federate) DeleteStatus(ctx context.Context, status *gtsmodel.Status) er return err } - // Drop any outgoing queued AP requests about / targeting - // this status, (stops queued likes, boosts, creates etc). - f.state.Workers.Delivery.Queue.Delete("ObjectID", status.URI) - f.state.Workers.Delivery.Queue.Delete("TargetID", status.URI) - - // Drop any incoming queued client messages about / targeting - // status, (stops processing of local origin data for status). - f.state.Workers.Client.Queue.Delete("TargetURI", status.URI) - - // Drop any incoming queued federator messages targeting status, - // (stops processing of remote origin data targeting this status). - f.state.Workers.Federator.Queue.Delete("TargetURI", 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) - } - // Wrap the status URI in a Delete activity. delete, err := f.converter.StatusToASDelete(ctx, status) if err != nil { diff --git a/internal/processing/workers/fromclientapi.go b/internal/processing/workers/fromclientapi.go index 701642032..63f2d5de7 100644 --- a/internal/processing/workers/fromclientapi.go +++ b/internal/processing/workers/fromclientapi.go @@ -633,10 +633,27 @@ func (p *clientAPI) DeleteStatus(ctx context.Context, cMsg *messages.FromClientA return gtserror.Newf("db error populating status: %w", err) } + // First perform the actual status deletion. if err := p.utils.wipeStatus(ctx, status, deleteAttachments); err != nil { log.Errorf(ctx, "error wiping status: %v", err) } + // Now status is deleted, first thing we do is drop any + // queued work relating to it ASAP to prevent wasted work. + + // Drop any outgoing queued AP requests about / targeting + // this status, (stops queued likes, boosts, creates etc). + p.state.Workers.Delivery.Queue.Delete("ObjectID", status.URI) + p.state.Workers.Delivery.Queue.Delete("TargetID", status.URI) + + // Drop any incoming queued client messages about / targeting + // status, (stops processing of local origin data for status). + p.state.Workers.Client.Queue.Delete("TargetURI", status.URI) + + // Drop any incoming queued federator messages targeting status, + // (stops processing of remote origin data targeting this status). + p.state.Workers.Federator.Queue.Delete("TargetURI", status.URI) + // Update stats for the origin account. if err := p.utils.decrementStatusesCount(ctx, cMsg.Origin); err != nil { log.Errorf(ctx, "error updating account stats: %v", err) @@ -672,14 +689,38 @@ func (p *clientAPI) DeleteAccount(ctx context.Context, cMsg *messages.FromClient originID = cMsg.Origin.ID } - if err := p.federate.DeleteAccount(ctx, cMsg.Target); err != nil { - log.Errorf(ctx, "error federating account delete: %v", err) - } + // Extract target account. + account := cMsg.Target + // First perform the actual account deletion. if err := p.account.Delete(ctx, cMsg.Target, originID); err != nil { log.Errorf(ctx, "error deleting account: %v", err) } + // Now account is deleted, first thing we do is drop any + // queued work relating to it ASAP to prevent wasted work. + + // Drop any outgoing queued AP requests to / from / targeting + // this account, (stops queued likes, boosts, creates etc). + p.state.Workers.Delivery.Queue.Delete("ActorID", account.URI) + p.state.Workers.Delivery.Queue.Delete("ObjectID", account.URI) + p.state.Workers.Delivery.Queue.Delete("TargetID", account.URI) + + // Drop any incoming queued client messages to / from this + // account, (stops processing of local origin data for acccount). + p.state.Workers.Client.Queue.Delete("Origin.ID", account.ID) + p.state.Workers.Client.Queue.Delete("Target.ID", account.ID) + p.state.Workers.Client.Queue.Delete("TargetURI", account.URI) + + // Drop any incoming queued federator messages to this account, + // (stops processing of remote origin data targeting this account). + p.state.Workers.Federator.Queue.Delete("Receiving.ID", account.ID) + p.state.Workers.Federator.Queue.Delete("TargetURI", account.URI) + + if err := p.federate.DeleteAccount(ctx, cMsg.Target); err != nil { + log.Errorf(ctx, "error federating account delete: %v", err) + } + return nil } diff --git a/internal/processing/workers/fromfediapi.go b/internal/processing/workers/fromfediapi.go index abfb2f7e0..0f89db2ea 100644 --- a/internal/processing/workers/fromfediapi.go +++ b/internal/processing/workers/fromfediapi.go @@ -19,10 +19,12 @@ package workers import ( "context" + "errors" "codeberg.org/gruf/go-kv" "codeberg.org/gruf/go-logger/v2/level" "github.com/superseriousbusiness/gotosocial/internal/ap" + "github.com/superseriousbusiness/gotosocial/internal/db" "github.com/superseriousbusiness/gotosocial/internal/federation/dereferencing" "github.com/superseriousbusiness/gotosocial/internal/gtserror" @@ -601,10 +603,35 @@ func (p *fediAPI) DeleteStatus(ctx context.Context, fMsg *messages.FromFediAPI) return gtserror.Newf("%T not parseable as *gtsmodel.Status", fMsg.GTSModel) } + // Try to populate status structs if possible, + // in order to more thoroughly remove them. + if err := p.state.DB.PopulateStatus( + ctx, status, + ); err != nil && !errors.Is(err, db.ErrNoEntries) { + return gtserror.Newf("db error populating status: %w", err) + } + + // First perform the actual status deletion. if err := p.utils.wipeStatus(ctx, status, deleteAttachments); err != nil { log.Errorf(ctx, "error wiping status: %v", err) } + // Now status is deleted, first thing we do is drop any + // queued work relating to it ASAP to prevent wasted work. + + // Drop any outgoing queued AP requests about / targeting + // this status, (stops queued likes, boosts, creates etc). + p.state.Workers.Delivery.Queue.Delete("ObjectID", status.URI) + p.state.Workers.Delivery.Queue.Delete("TargetID", status.URI) + + // Drop any incoming queued client messages about / targeting + // status, (stops processing of local origin data for status). + p.state.Workers.Client.Queue.Delete("TargetURI", status.URI) + + // Drop any incoming queued federator messages targeting status, + // (stops processing of remote origin data targeting this status). + p.state.Workers.Federator.Queue.Delete("TargetURI", status.URI) + // Update stats for the remote account. if err := p.utils.decrementStatusesCount(ctx, fMsg.Requesting); err != nil { log.Errorf(ctx, "error updating account stats: %v", err) @@ -625,9 +652,28 @@ func (p *fediAPI) DeleteAccount(ctx context.Context, fMsg *messages.FromFediAPI) return gtserror.Newf("%T not parseable as *gtsmodel.Account", fMsg.GTSModel) } + // First perform the actual account deletion. if err := p.account.Delete(ctx, account, account.ID); err != nil { log.Errorf(ctx, "error deleting account: %v", err) } + // Now account is deleted, first thing we do is drop any + // queued work relating to it ASAP to prevent wasted work. + + // Drop any outgoing queued AP requests to / from / targeting + // this account, (stops queued likes, boosts, creates etc). + p.state.Workers.Delivery.Queue.Delete("ObjectID", account.URI) + p.state.Workers.Delivery.Queue.Delete("TargetID", account.URI) + + // Drop any incoming queued client messages to / from this + // account, (stops processing of local origin data for acccount). + p.state.Workers.Client.Queue.Delete("Target.ID", account.ID) + p.state.Workers.Client.Queue.Delete("TargetURI", account.URI) + + // Drop any incoming queued federator messages to this account, + // (stops processing of remote origin data targeting this account). + p.state.Workers.Federator.Queue.Delete("Requesting.ID", account.ID) + p.state.Workers.Federator.Queue.Delete("TargetURI", account.URI) + return nil }