move all queue purging to the processor workers

This commit is contained in:
kim 2024-04-24 12:54:51 +01:00
parent 4729c92dad
commit 5f81198328
4 changed files with 90 additions and 72 deletions

View file

@ -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,

View file

@ -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 {

View file

@ -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
}

View file

@ -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
}