diff options
Diffstat (limited to 'internal')
63 files changed, 1401 insertions, 757 deletions
diff --git a/internal/api/client/statuses/status_test.go b/internal/api/client/statuses/status_test.go index 6d0710ef0..a979f0c00 100644 --- a/internal/api/client/statuses/status_test.go +++ b/internal/api/client/statuses/status_test.go @@ -72,7 +72,6 @@ func (suite *StatusStandardTestSuite) SetupSuite() { func (suite *StatusStandardTestSuite) SetupTest() { suite.state.Caches.Init() - testrig.StartNoopWorkers(&suite.state) testrig.InitTestConfig() testrig.InitTestLog() @@ -98,6 +97,8 @@ func (suite *StatusStandardTestSuite) SetupTest() { suite.emailSender = testrig.NewEmailSender("../../../../web/template/", nil) suite.processor = testrig.NewTestProcessor(&suite.state, suite.federator, suite.emailSender, suite.mediaManager) suite.statusModule = statuses.New(suite.processor) + + testrig.StartWorkers(&suite.state, suite.processor.Workers()) } func (suite *StatusStandardTestSuite) TearDownTest() { diff --git a/internal/api/client/statuses/statusdelete_test.go b/internal/api/client/statuses/statusdelete_test.go index f94a0bb08..6f62ef45f 100644 --- a/internal/api/client/statuses/statusdelete_test.go +++ b/internal/api/client/statuses/statusdelete_test.go @@ -19,7 +19,7 @@ import ( "encoding/json" "errors" "fmt" - "io/ioutil" + "io" "net/http" "net/http/httptest" "strings" @@ -69,7 +69,7 @@ func (suite *StatusDeleteTestSuite) TestPostDelete() { result := recorder.Result() defer result.Body.Close() - b, err := ioutil.ReadAll(result.Body) + b, err := io.ReadAll(result.Body) suite.NoError(err) statusReply := &apimodel.Status{} diff --git a/internal/federation/dereferencing/account.go b/internal/federation/dereferencing/account.go index e8d32f58a..94df9538a 100644 --- a/internal/federation/dereferencing/account.go +++ b/internal/federation/dereferencing/account.go @@ -104,7 +104,7 @@ func (d *Dereferencer) GetAccountByURI(ctx context.Context, requestUser string, if accountable != nil { // This account was updated, enqueue re-dereference featured posts. - d.state.Workers.Federator.MustEnqueueCtx(ctx, func(ctx context.Context) { + d.state.Workers.Dereference.Queue.Push(func(ctx context.Context) { if err := d.dereferenceAccountFeatured(ctx, requestUser, account); err != nil { log.Errorf(ctx, "error fetching account featured collection: %v", err) } @@ -201,7 +201,7 @@ func (d *Dereferencer) GetAccountByUsernameDomain(ctx context.Context, requestUs if accountable != nil { // This account was updated, enqueue re-dereference featured posts. - d.state.Workers.Federator.MustEnqueueCtx(ctx, func(ctx context.Context) { + d.state.Workers.Dereference.Queue.Push(func(ctx context.Context) { if err := d.dereferenceAccountFeatured(ctx, requestUser, account); err != nil { log.Errorf(ctx, "error fetching account featured collection: %v", err) } @@ -322,7 +322,7 @@ func (d *Dereferencer) RefreshAccount( if accountable != nil { // This account was updated, enqueue re-dereference featured posts. - d.state.Workers.Federator.MustEnqueueCtx(ctx, func(ctx context.Context) { + d.state.Workers.Dereference.Queue.Push(func(ctx context.Context) { if err := d.dereferenceAccountFeatured(ctx, requestUser, latest); err != nil { log.Errorf(ctx, "error fetching account featured collection: %v", err) } @@ -362,7 +362,7 @@ func (d *Dereferencer) RefreshAccountAsync( } // Enqueue a worker function to enrich this account async. - d.state.Workers.Federator.MustEnqueueCtx(ctx, func(ctx context.Context) { + d.state.Workers.Dereference.Queue.Push(func(ctx context.Context) { latest, accountable, err := d.enrichAccountSafely(ctx, requestUser, uri, account, accountable) if err != nil { log.Errorf(ctx, "error enriching remote account: %v", err) diff --git a/internal/federation/dereferencing/status.go b/internal/federation/dereferencing/status.go index 769539a2f..bd50a08fd 100644 --- a/internal/federation/dereferencing/status.go +++ b/internal/federation/dereferencing/status.go @@ -255,7 +255,7 @@ func (d *Dereferencer) RefreshStatusAsync( } // Enqueue a worker function to re-fetch this status entirely async. - d.state.Workers.Federator.MustEnqueueCtx(ctx, func(ctx context.Context) { + d.state.Workers.Dereference.Queue.Push(func(ctx context.Context) { latest, statusable, _, err := d.enrichStatusSafely(ctx, requestUser, uri, diff --git a/internal/federation/dereferencing/thread.go b/internal/federation/dereferencing/thread.go index e528581c9..ed2c1a43f 100644 --- a/internal/federation/dereferencing/thread.go +++ b/internal/federation/dereferencing/thread.go @@ -56,14 +56,14 @@ func (d *Dereferencer) dereferenceThread( } // Enqueue dereferencing remaining status thread, (children), asychronously . - d.state.Workers.Federator.MustEnqueueCtx(ctx, func(ctx context.Context) { + d.state.Workers.Dereference.Queue.Push(func(ctx context.Context) { if err := d.DereferenceStatusDescendants(ctx, requestUser, uri, statusable); err != nil { log.Error(ctx, err) } }) } else { // This is an existing status, dereference the WHOLE thread asynchronously. - d.state.Workers.Federator.MustEnqueueCtx(ctx, func(ctx context.Context) { + d.state.Workers.Dereference.Queue.Push(func(ctx context.Context) { if err := d.DereferenceStatusAncestors(ctx, requestUser, status); err != nil { log.Error(ctx, err) } diff --git a/internal/federation/federatingdb/accept.go b/internal/federation/federatingdb/accept.go index 50a7c2db1..e26e5955b 100644 --- a/internal/federation/federatingdb/accept.go +++ b/internal/federation/federatingdb/accept.go @@ -89,13 +89,12 @@ func (f *federatingDB) Accept(ctx context.Context, accept vocab.ActivityStreamsA return err } - // Process side effects asynchronously. - f.state.Workers.EnqueueFediAPI(ctx, messages.FromFediAPI{ - APObjectType: ap.ActivityFollow, - APActivityType: ap.ActivityAccept, - GTSModel: follow, - ReceivingAccount: receivingAcct, - RequestingAccount: requestingAcct, + f.state.Workers.Federator.Queue.Push(&messages.FromFediAPI{ + APObjectType: ap.ActivityFollow, + APActivityType: ap.ActivityAccept, + GTSModel: follow, + Receiving: receivingAcct, + Requesting: requestingAcct, }) } @@ -138,13 +137,12 @@ func (f *federatingDB) Accept(ctx context.Context, accept vocab.ActivityStreamsA return err } - // Process side effects asynchronously. - f.state.Workers.EnqueueFediAPI(ctx, messages.FromFediAPI{ - APObjectType: ap.ActivityFollow, - APActivityType: ap.ActivityAccept, - GTSModel: follow, - ReceivingAccount: receivingAcct, - RequestingAccount: requestingAcct, + f.state.Workers.Federator.Queue.Push(&messages.FromFediAPI{ + APObjectType: ap.ActivityFollow, + APActivityType: ap.ActivityAccept, + GTSModel: follow, + Receiving: receivingAcct, + Requesting: requestingAcct, }) continue diff --git a/internal/federation/federatingdb/announce.go b/internal/federation/federatingdb/announce.go index 2f5950a30..3a3b91236 100644 --- a/internal/federation/federatingdb/announce.go +++ b/internal/federation/federatingdb/announce.go @@ -81,12 +81,12 @@ func (f *federatingDB) Announce(ctx context.Context, announce vocab.ActivityStre } // This is a new boost. Process side effects asynchronously. - f.state.Workers.EnqueueFediAPI(ctx, messages.FromFediAPI{ - APObjectType: ap.ActivityAnnounce, - APActivityType: ap.ActivityCreate, - GTSModel: boost, - ReceivingAccount: receivingAcct, - RequestingAccount: requestingAcct, + f.state.Workers.Federator.Queue.Push(&messages.FromFediAPI{ + APObjectType: ap.ActivityAnnounce, + APActivityType: ap.ActivityCreate, + GTSModel: boost, + Receiving: receivingAcct, + Requesting: requestingAcct, }) return nil diff --git a/internal/federation/federatingdb/announce_test.go b/internal/federation/federatingdb/announce_test.go index 8dd5ce9da..2833c04c4 100644 --- a/internal/federation/federatingdb/announce_test.go +++ b/internal/federation/federatingdb/announce_test.go @@ -19,6 +19,7 @@ package federatingdb_test import ( "testing" + "time" "github.com/stretchr/testify/suite" "github.com/superseriousbusiness/activity/streams/vocab" @@ -42,7 +43,7 @@ func (suite *AnnounceTestSuite) TestNewAnnounce() { suite.NoError(err) // should be a message heading to the processor now, which we can intercept here - msg := <-suite.fromFederator + msg, _ := suite.getFederatorMsg(5 * time.Second) suite.Equal(ap.ActivityAnnounce, msg.APObjectType) suite.Equal(ap.ActivityCreate, msg.APActivityType) @@ -69,7 +70,7 @@ func (suite *AnnounceTestSuite) TestAnnounceTwice() { suite.NoError(err) // should be a message heading to the processor now, which we can intercept here - msg := <-suite.fromFederator + msg, _ := suite.getFederatorMsg(5 * time.Second) suite.Equal(ap.ActivityAnnounce, msg.APObjectType) suite.Equal(ap.ActivityCreate, msg.APActivityType) boost, ok := msg.GTSModel.(*gtsmodel.Status) @@ -94,7 +95,8 @@ func (suite *AnnounceTestSuite) TestAnnounceTwice() { // since this is a repeat announce with the same URI, just delivered to a different inbox, // we should have nothing in the messages channel... - suite.Empty(suite.fromFederator) + _, ok = suite.getFederatorMsg(time.Second) + suite.False(ok) } func TestAnnounceTestSuite(t *testing.T) { 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 diff --git a/internal/federation/federatingdb/create_test.go b/internal/federation/federatingdb/create_test.go index 5f80812bf..fffee1432 100644 --- a/internal/federation/federatingdb/create_test.go +++ b/internal/federation/federatingdb/create_test.go @@ -21,6 +21,7 @@ import ( "context" "encoding/json" "testing" + "time" "github.com/stretchr/testify/suite" "github.com/superseriousbusiness/activity/streams" @@ -48,10 +49,10 @@ func (suite *CreateTestSuite) TestCreateNote() { suite.NoError(err) // should be a message heading to the processor now, which we can intercept here - msg := <-suite.fromFederator + msg, _ := suite.getFederatorMsg(5 * time.Second) suite.Equal(ap.ObjectNote, msg.APObjectType) suite.Equal(ap.ActivityCreate, msg.APActivityType) - suite.Equal(note, msg.APObjectModel) + suite.Equal(note, msg.APObject) } func (suite *CreateTestSuite) TestCreateNoteForward() { @@ -79,15 +80,15 @@ func (suite *CreateTestSuite) TestCreateNoteForward() { suite.NoError(err) // should be a message heading to the processor now, which we can intercept here - msg := <-suite.fromFederator + msg, _ := suite.getFederatorMsg(5 * time.Second) suite.Equal(ap.ObjectNote, msg.APObjectType) suite.Equal(ap.ActivityCreate, msg.APActivityType) // nothing should be set as the model since this is a forward - suite.Nil(msg.APObjectModel) + suite.Nil(msg.APObject) // but we should have a uri set - suite.Equal("http://example.org/users/Some_User/statuses/afaba698-5740-4e32-a702-af61aa543bc1", msg.APIri.String()) + suite.Equal("http://example.org/users/Some_User/statuses/afaba698-5740-4e32-a702-af61aa543bc1", msg.APIRI.String()) } func (suite *CreateTestSuite) TestCreateFlag1() { @@ -120,7 +121,7 @@ func (suite *CreateTestSuite) TestCreateFlag1() { } // should be a message heading to the processor now, which we can intercept here - msg := <-suite.fromFederator + msg, _ := suite.getFederatorMsg(5 * time.Second) suite.Equal(ap.ActivityFlag, msg.APObjectType) suite.Equal(ap.ActivityCreate, msg.APActivityType) 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 } diff --git a/internal/federation/federatingdb/federatingdb_test.go b/internal/federation/federatingdb/federatingdb_test.go index 0f227164d..360094887 100644 --- a/internal/federation/federatingdb/federatingdb_test.go +++ b/internal/federation/federatingdb/federatingdb_test.go @@ -19,6 +19,7 @@ package federatingdb_test import ( "context" + "time" "github.com/stretchr/testify/suite" "github.com/superseriousbusiness/gotosocial/internal/db" @@ -34,11 +35,10 @@ import ( type FederatingDBTestSuite struct { suite.Suite - db db.DB - tc *typeutils.Converter - fromFederator chan messages.FromFediAPI - federatingDB federatingdb.DB - state state.State + db db.DB + tc *typeutils.Converter + federatingDB federatingdb.DB + state state.State testTokens map[string]*gtsmodel.Token testClients map[string]*gtsmodel.Client @@ -51,6 +51,13 @@ type FederatingDBTestSuite struct { testActivities map[string]testrig.ActivityWithSignature } +func (suite *FederatingDBTestSuite) getFederatorMsg(timeout time.Duration) (*messages.FromFediAPI, bool) { + ctx := context.Background() + ctx, cncl := context.WithTimeout(ctx, timeout) + defer cncl() + return suite.state.Workers.Federator.Queue.PopCtx(ctx) +} + func (suite *FederatingDBTestSuite) SetupSuite() { suite.testTokens = testrig.NewTestTokens() suite.testClients = testrig.NewTestClients() @@ -69,13 +76,6 @@ func (suite *FederatingDBTestSuite) SetupTest() { suite.state.Caches.Init() testrig.StartNoopWorkers(&suite.state) - suite.fromFederator = make(chan messages.FromFediAPI, 10) - suite.state.Workers.EnqueueFediAPI = func(ctx context.Context, msgs ...messages.FromFediAPI) { - for _, msg := range msgs { - suite.fromFederator <- msg - } - } - suite.db = testrig.NewTestDB(&suite.state) suite.testActivities = testrig.NewTestActivities(suite.testAccounts) @@ -96,13 +96,6 @@ func (suite *FederatingDBTestSuite) SetupTest() { func (suite *FederatingDBTestSuite) TearDownTest() { testrig.StandardDBTeardown(suite.db) testrig.StopWorkers(&suite.state) - for suite.fromFederator != nil { - select { - case <-suite.fromFederator: - default: - return - } - } } func createTestContext(receivingAccount *gtsmodel.Account, requestingAccount *gtsmodel.Account) context.Context { diff --git a/internal/federation/federatingdb/move.go b/internal/federation/federatingdb/move.go index 2e8049e08..59dc2529c 100644 --- a/internal/federation/federatingdb/move.go +++ b/internal/federation/federatingdb/move.go @@ -170,12 +170,12 @@ func (f *federatingDB) Move(ctx context.Context, move vocab.ActivityStreamsMove) // We had a Move already or stored a new Move. // Pass back to a worker for async processing. - f.state.Workers.EnqueueFediAPI(ctx, messages.FromFediAPI{ - APObjectType: ap.ObjectProfile, - APActivityType: ap.ActivityMove, - GTSModel: stubMove, - RequestingAccount: requestingAcct, - ReceivingAccount: receivingAcct, + f.state.Workers.Federator.Queue.Push(&messages.FromFediAPI{ + APObjectType: ap.ObjectProfile, + APActivityType: ap.ActivityMove, + GTSModel: stubMove, + Requesting: requestingAcct, + Receiving: receivingAcct, }) return nil diff --git a/internal/federation/federatingdb/move_test.go b/internal/federation/federatingdb/move_test.go index 006dcf0dc..3e35dc97a 100644 --- a/internal/federation/federatingdb/move_test.go +++ b/internal/federation/federatingdb/move_test.go @@ -27,7 +27,6 @@ import ( "github.com/superseriousbusiness/activity/streams/vocab" "github.com/superseriousbusiness/gotosocial/internal/ap" "github.com/superseriousbusiness/gotosocial/internal/gtsmodel" - "github.com/superseriousbusiness/gotosocial/internal/messages" ) type MoveTestSuite struct { @@ -78,13 +77,7 @@ func (suite *MoveTestSuite) TestMove() { suite.move(receivingAcct, requestingAcct, moveStr1) // Should be a message heading to the processor. - var msg messages.FromFediAPI - select { - case msg = <-suite.fromFederator: - // Fine. - case <-time.After(5 * time.Second): - suite.FailNow("", "timeout waiting for suite.fromFederator") - } + msg, _ := suite.getFederatorMsg(5 * time.Second) suite.Equal(ap.ObjectProfile, msg.APObjectType) suite.Equal(ap.ActivityMove, msg.APActivityType) @@ -101,12 +94,7 @@ func (suite *MoveTestSuite) TestMove() { // Should be a message heading to the processor // since this is just a straight up retry. - select { - case msg = <-suite.fromFederator: - // Fine. - case <-time.After(5 * time.Second): - suite.FailNow("", "timeout waiting for suite.fromFederator") - } + msg, _ = suite.getFederatorMsg(5 * time.Second) suite.Equal(ap.ObjectProfile, msg.APObjectType) suite.Equal(ap.ActivityMove, msg.APActivityType) @@ -126,12 +114,7 @@ func (suite *MoveTestSuite) TestMove() { // Should be a message heading to the processor // since this is just a retry with a different ID. - select { - case msg = <-suite.fromFederator: - // Fine. - case <-time.After(5 * time.Second): - suite.FailNow("", "timeout waiting for suite.fromFederator") - } + msg, _ = suite.getFederatorMsg(5 * time.Second) suite.Equal(ap.ObjectProfile, msg.APObjectType) suite.Equal(ap.ActivityMove, msg.APActivityType) } diff --git a/internal/federation/federatingdb/reject_test.go b/internal/federation/federatingdb/reject_test.go index d4c537a92..f51ffaf56 100644 --- a/internal/federation/federatingdb/reject_test.go +++ b/internal/federation/federatingdb/reject_test.go @@ -81,7 +81,8 @@ func (suite *RejectTestSuite) TestRejectFollowRequest() { suite.NoError(err) // there should be nothing in the federator channel since nothing needs to be passed - suite.Empty(suite.fromFederator) + _, ok := suite.getFederatorMsg(time.Second) + suite.False(ok) // the follow request should not be in the database anymore -- it's been rejected err = suite.db.GetByID(ctx, fr.ID, >smodel.FollowRequest{}) diff --git a/internal/federation/federatingdb/update.go b/internal/federation/federatingdb/update.go index 733abba0d..2f00e0867 100644 --- a/internal/federation/federatingdb/update.go +++ b/internal/federation/federatingdb/update.go @@ -98,13 +98,13 @@ func (f *federatingDB) updateAccountable(ctx context.Context, receivingAcct *gts // was delivered along with the Update, for further asynchronous // updating of eg., avatar/header, emojis, etc. The actual db // inserts/updates will take place there. - f.state.Workers.EnqueueFediAPI(ctx, messages.FromFediAPI{ - APObjectType: ap.ObjectProfile, - APActivityType: ap.ActivityUpdate, - GTSModel: requestingAcct, - APObjectModel: accountable, - ReceivingAccount: receivingAcct, - RequestingAccount: requestingAcct, + f.state.Workers.Federator.Queue.Push(&messages.FromFediAPI{ + APObjectType: ap.ObjectProfile, + APActivityType: ap.ActivityUpdate, + GTSModel: requestingAcct, + APObject: accountable, + Receiving: receivingAcct, + Requesting: requestingAcct, }) return nil @@ -155,13 +155,13 @@ func (f *federatingDB) updateStatusable(ctx context.Context, receivingAcct *gtsm // Queue an UPDATE NOTE activity to our fedi API worker, // this will handle necessary database insertions, etc. - f.state.Workers.EnqueueFediAPI(ctx, messages.FromFediAPI{ - APObjectType: ap.ObjectNote, - APActivityType: ap.ActivityUpdate, - GTSModel: status, // original status - APObjectModel: (ap.Statusable)(statusable), - ReceivingAccount: receivingAcct, - RequestingAccount: requestingAcct, + f.state.Workers.Federator.Queue.Push(&messages.FromFediAPI{ + APObjectType: ap.ObjectNote, + APActivityType: ap.ActivityUpdate, + GTSModel: status, // original status + APObject: (ap.Statusable)(statusable), + Receiving: receivingAcct, + Requesting: requestingAcct, }) return nil diff --git a/internal/httpclient/client.go b/internal/httpclient/client.go index 31c6df7d0..ed0949c9c 100644 --- a/internal/httpclient/client.go +++ b/internal/httpclient/client.go @@ -59,9 +59,27 @@ var ( // configuration values passed to initialized http.Transport{} // and http.Client{}, along with httpclient.Client{} specific. type Config struct { - // MaxOpenConnsPerHost limits the max number of open connections to a host. + + // MaxOpenConnsPerHost limits the max + // number of open connections to a host. MaxOpenConnsPerHost int + // AllowRanges allows outgoing + // communications to given IP nets. + AllowRanges []netip.Prefix + + // BlockRanges blocks outgoing + // communiciations to given IP nets. + BlockRanges []netip.Prefix + + // TLSInsecureSkipVerify can be set to true to + // skip validation of remote TLS certificates. + // + // THIS SHOULD BE USED FOR TESTING ONLY, IF YOU + // TURN THIS ON WHILE RUNNING IN PRODUCTION YOU + // ARE LEAVING YOUR SERVER WIDE OPEN TO ATTACKS! + TLSInsecureSkipVerify bool + // MaxIdleConns: see http.Transport{}.MaxIdleConns. MaxIdleConns int @@ -79,20 +97,6 @@ type Config struct { // DisableCompression: see http.Transport{}.DisableCompression. DisableCompression bool - - // AllowRanges allows outgoing communications to given IP nets. - AllowRanges []netip.Prefix - - // BlockRanges blocks outgoing communiciations to given IP nets. - BlockRanges []netip.Prefix - - // TLSInsecureSkipVerify can be set to true to - // skip validation of remote TLS certificates. - // - // THIS SHOULD BE USED FOR TESTING ONLY, IF YOU - // TURN THIS ON WHILE RUNNING IN PRODUCTION YOU - // ARE LEAVING YOUR SERVER WIDE OPEN TO ATTACKS! - TLSInsecureSkipVerify bool } // Client wraps an underlying http.Client{} to provide the following: @@ -138,7 +142,8 @@ func New(cfg Config) *Client { cfg.MaxBodySize = int64(40 * bytesize.MiB) } - // Protect dialer with IP range sanitizer. + // Protect the dialer + // with IP range sanitizer. d.Control = (&Sanitizer{ Allow: cfg.AllowRanges, Block: cfg.BlockRanges, @@ -148,7 +153,7 @@ func New(cfg Config) *Client { c.client.Timeout = cfg.Timeout c.bodyMax = cfg.MaxBodySize - // Prepare TLS config for transport. + // Prepare transport TLS config. tlsClientConfig := &tls.Config{ InsecureSkipVerify: cfg.TLSInsecureSkipVerify, //nolint:gosec } diff --git a/internal/media/manager.go b/internal/media/manager.go index 61f4e73dd..73494881c 100644 --- a/internal/media/manager.go +++ b/internal/media/manager.go @@ -432,8 +432,8 @@ func (m *Manager) ProcessEmoji( return nil, err } - // Attempt to add this emoji processing item to the worker queue. - _ = m.state.Workers.Media.MustEnqueueCtx(ctx, emoji.Process) + // Attempt to add emoji item to the worker queue. + m.state.Workers.Media.Queue.Push(emoji.Process) return emoji, nil } diff --git a/internal/media/processingemoji.go b/internal/media/processingemoji.go index dfe7c1883..b62c4f76e 100644 --- a/internal/media/processingemoji.go +++ b/internal/media/processingemoji.go @@ -67,7 +67,7 @@ func (p *ProcessingEmoji) LoadEmoji(ctx context.Context) (*gtsmodel.Emoji, error // Provided context was cancelled, e.g. request cancelled // early. Queue this item for asynchronous processing. log.Warnf(ctx, "reprocessing emoji %s after canceled ctx", p.emoji.ID) - go p.mgr.state.Workers.Media.Enqueue(p.Process) + p.mgr.state.Workers.Media.Queue.Push(p.Process) } return nil, err diff --git a/internal/media/processingmedia.go b/internal/media/processingmedia.go index ffedc3208..5975de859 100644 --- a/internal/media/processingmedia.go +++ b/internal/media/processingmedia.go @@ -27,7 +27,7 @@ import ( errorsv2 "codeberg.org/gruf/go-errors/v2" "codeberg.org/gruf/go-runners" - "codeberg.org/superseriousbusiness/exif-terminator" + terminator "codeberg.org/superseriousbusiness/exif-terminator" "github.com/disintegration/imaging" "github.com/h2non/filetype" "github.com/superseriousbusiness/gotosocial/internal/gtserror" @@ -82,7 +82,7 @@ func (p *ProcessingMedia) LoadAttachment(ctx context.Context) (*gtsmodel.MediaAt // asynchronous processing, which will // use a background context. log.Warnf(ctx, "reprocessing media %s after canceled ctx", p.media.ID) - go p.mgr.state.Workers.Media.Enqueue(p.Process) + p.mgr.state.Workers.Media.Queue.Push(p.Process) } // Media could not be retrieved FULLY, diff --git a/internal/messages/messages.go b/internal/messages/messages.go index 32cb5fbba..c5488d586 100644 --- a/internal/messages/messages.go +++ b/internal/messages/messages.go @@ -20,25 +20,84 @@ package messages import ( "net/url" + "codeberg.org/gruf/go-structr" "github.com/superseriousbusiness/gotosocial/internal/gtsmodel" ) -// FromClientAPI wraps a message that travels from the client API into the processor. +// FromClientAPI wraps a message that +// travels from the client API into the processor. type FromClientAPI struct { - APObjectType string + + // APObjectType ... + APObjectType string + + // APActivityType ... APActivityType string - GTSModel interface{} - OriginAccount *gtsmodel.Account - TargetAccount *gtsmodel.Account + + // Optional GTS database model + // of the Activity / Object. + GTSModel interface{} + + // Targeted object URI. + TargetURI string + + // Origin is the account that + // this message originated from. + Origin *gtsmodel.Account + + // Target is the account that + // this message is targeting. + Target *gtsmodel.Account +} + +// ClientMsgIndices defines queue indices this +// message type should be accessible / stored under. +func ClientMsgIndices() []structr.IndexConfig { + return []structr.IndexConfig{ + {Fields: "TargetURI", Multiple: true}, + {Fields: "Origin.ID", Multiple: true}, + {Fields: "Target.ID", Multiple: true}, + } } -// FromFediAPI wraps a message that travels from the federating API into the processor. +// FromFediAPI wraps a message that +// travels from the federating API into the processor. type FromFediAPI struct { - APObjectType string - APActivityType string - APIri *url.URL - APObjectModel interface{} // Optional AP model of the Object of the Activity. Should be Accountable or Statusable. - GTSModel interface{} // Optional GTS model of the Activity or Object. - RequestingAccount *gtsmodel.Account // Remote account that posted this Activity to the inbox. - ReceivingAccount *gtsmodel.Account // Local account which owns the inbox that this Activity was posted to. + + // APObjectType ... + APObjectType string + + // APActivityType ... + APActivityType string + + // Optional ActivityPub ID (IRI) + // and / or model of Activity / Object. + APIRI *url.URL + APObject interface{} + + // Optional GTS database model + // of the Activity / Object. + GTSModel interface{} + + // Targeted object URI. + TargetURI string + + // Remote account that posted + // this Activity to the inbox. + Requesting *gtsmodel.Account + + // Local account which owns the inbox + // that this Activity was posted to. + Receiving *gtsmodel.Account +} + +// FederatorMsgIndices defines queue indices this +// message type should be accessible / stored under. +func FederatorMsgIndices() []structr.IndexConfig { + return []structr.IndexConfig{ + {Fields: "APIRI", Multiple: true}, + {Fields: "TargetURI", Multiple: true}, + {Fields: "Requesting.ID", Multiple: true}, + {Fields: "Receiving.ID", Multiple: true}, + } } diff --git a/internal/processing/account/account_test.go b/internal/processing/account/account_test.go index 244fc89d6..10d5f91e1 100644 --- a/internal/processing/account/account_test.go +++ b/internal/processing/account/account_test.go @@ -19,6 +19,7 @@ package account_test import ( "context" + "time" "github.com/stretchr/testify/suite" "github.com/superseriousbusiness/gotosocial/internal/db" @@ -48,7 +49,6 @@ type AccountStandardTestSuite struct { state state.State mediaManager *media.Manager oauthServer oauth.Server - fromClientAPIChan chan messages.FromClientAPI transportController transport.Controller federator *federation.Federator emailSender email.Sender @@ -68,6 +68,13 @@ type AccountStandardTestSuite struct { accountProcessor account.Processor } +func (suite *AccountStandardTestSuite) getClientMsg(timeout time.Duration) (*messages.FromClientAPI, bool) { + ctx := context.Background() + ctx, cncl := context.WithTimeout(ctx, timeout) + defer cncl() + return suite.state.Workers.Client.Queue.PopCtx(ctx) +} + func (suite *AccountStandardTestSuite) SetupSuite() { suite.testTokens = testrig.NewTestTokens() suite.testClients = testrig.NewTestClients() @@ -101,13 +108,6 @@ func (suite *AccountStandardTestSuite) SetupTest() { suite.mediaManager = testrig.NewTestMediaManager(&suite.state) suite.oauthServer = testrig.NewTestOauthServer(suite.db) - suite.fromClientAPIChan = make(chan messages.FromClientAPI, 100) - suite.state.Workers.EnqueueClientAPI = func(ctx context.Context, msgs ...messages.FromClientAPI) { - for _, msg := range msgs { - suite.fromClientAPIChan <- msg - } - } - suite.transportController = testrig.NewTestTransportController(&suite.state, testrig.NewMockHTTPClient(nil, "../../../testrig/media")) suite.federator = testrig.NewTestFederator(&suite.state, suite.transportController, suite.mediaManager) suite.sentEmails = make(map[string]string) diff --git a/internal/processing/account/block.go b/internal/processing/account/block.go index 49f81557d..d3904bffa 100644 --- a/internal/processing/account/block.go +++ b/internal/processing/account/block.go @@ -83,16 +83,16 @@ func (p *Processor) BlockCreate(ctx context.Context, requestingAccount *gtsmodel } // Process block side effects (federation etc). - msgs = append(msgs, messages.FromClientAPI{ + msgs = append(msgs, &messages.FromClientAPI{ APObjectType: ap.ActivityBlock, APActivityType: ap.ActivityCreate, GTSModel: block, - OriginAccount: requestingAccount, - TargetAccount: targetAccount, + Origin: requestingAccount, + Target: targetAccount, }) // Batch queue accreted client api messages. - p.state.Workers.EnqueueClientAPI(ctx, msgs...) + p.state.Workers.Client.Queue.Push(msgs...) return p.RelationshipGet(ctx, requestingAccount, targetAccountID) } @@ -120,12 +120,12 @@ func (p *Processor) BlockRemove(ctx context.Context, requestingAccount *gtsmodel existingBlock.TargetAccount = targetAccount // Process block removal side effects (federation etc). - p.state.Workers.EnqueueClientAPI(ctx, messages.FromClientAPI{ + p.state.Workers.Client.Queue.Push(&messages.FromClientAPI{ APObjectType: ap.ActivityBlock, APActivityType: ap.ActivityUndo, GTSModel: existingBlock, - OriginAccount: requestingAccount, - TargetAccount: targetAccount, + Origin: requestingAccount, + Target: targetAccount, }) return p.RelationshipGet(ctx, requestingAccount, targetAccountID) diff --git a/internal/processing/account/create.go b/internal/processing/account/create.go index 12b2d5e57..761165356 100644 --- a/internal/processing/account/create.go +++ b/internal/processing/account/create.go @@ -126,11 +126,11 @@ func (p *Processor) Create( // There are side effects for creating a new account // (confirmation emails etc), perform these async. - p.state.Workers.EnqueueClientAPI(ctx, messages.FromClientAPI{ + p.state.Workers.Client.Queue.Push(&messages.FromClientAPI{ APObjectType: ap.ObjectProfile, APActivityType: ap.ActivityCreate, GTSModel: user, - OriginAccount: user.Account, + Origin: user.Account, }) return user, nil diff --git a/internal/processing/account/delete.go b/internal/processing/account/delete.go index a900c566d..3f051edf0 100644 --- a/internal/processing/account/delete.go +++ b/internal/processing/account/delete.go @@ -102,16 +102,13 @@ func (p *Processor) Delete( // and the above Delete function will be called afterwards from the processor, to clear // out the account's bits and bobs, and stubbify it. func (p *Processor) DeleteSelf(ctx context.Context, account *gtsmodel.Account) gtserror.WithCode { - fromClientAPIMessage := messages.FromClientAPI{ + // Process the delete side effects asynchronously. + p.state.Workers.Client.Queue.Push(&messages.FromClientAPI{ APObjectType: ap.ActorPerson, APActivityType: ap.ActivityDelete, - OriginAccount: account, - TargetAccount: account, - } - - // Process the delete side effects asynchronously. - p.state.Workers.EnqueueClientAPI(ctx, fromClientAPIMessage) - + Origin: account, + Target: account, + }) return nil } @@ -193,7 +190,8 @@ func (p *Processor) deleteAccountFollows(ctx context.Context, account *gtsmodel. var ( // Use this slice to batch unfollow messages. - msgs = []messages.FromClientAPI{} + msgs = []*messages.FromClientAPI{} + // To avoid checking if account is local over + over // inside the subsequent loops, just generate static // side effects function once now. @@ -214,7 +212,7 @@ func (p *Processor) deleteAccountFollows(ctx context.Context, account *gtsmodel. } if msg := unfollowSideEffects(ctx, account, follow); msg != nil { // There was a side effect to process. - msgs = append(msgs, *msg) + msgs = append(msgs, msg) } } @@ -244,13 +242,13 @@ func (p *Processor) deleteAccountFollows(ctx context.Context, account *gtsmodel. if msg := unfollowSideEffects(ctx, account, follow); msg != nil { // There was a side effect to process. - msgs = append(msgs, *msg) + msgs = append(msgs, msg) } } // Process accreted messages in serial. for _, msg := range msgs { - if err := p.state.Workers.ProcessFromClientAPI(ctx, msg); err != nil { + if err := p.state.Workers.Client.Process(ctx, msg); err != nil { log.Errorf( ctx, "error processing %s of %s during Delete of account %s: %v", @@ -306,8 +304,8 @@ func (p *Processor) unfollowSideEffectsFunc(local bool) func( APObjectType: ap.ActivityFollow, APActivityType: ap.ActivityUndo, GTSModel: follow, - OriginAccount: account, - TargetAccount: follow.TargetAccount, + Origin: account, + Target: follow.TargetAccount, } } } @@ -337,7 +335,7 @@ func (p *Processor) deleteAccountStatuses( statuses []*gtsmodel.Status err error maxID string - msgs = []messages.FromClientAPI{} + msgs = []*messages.FromClientAPI{} ) statusLoop: @@ -404,29 +402,29 @@ statusLoop: continue } - msgs = append(msgs, messages.FromClientAPI{ + msgs = append(msgs, &messages.FromClientAPI{ APObjectType: ap.ActivityAnnounce, APActivityType: ap.ActivityUndo, GTSModel: status, - OriginAccount: boost.Account, - TargetAccount: account, + Origin: boost.Account, + Target: account, }) } // Now prepare to Delete status. - msgs = append(msgs, messages.FromClientAPI{ + msgs = append(msgs, &messages.FromClientAPI{ APObjectType: ap.ObjectNote, APActivityType: ap.ActivityDelete, GTSModel: status, - OriginAccount: account, - TargetAccount: account, + Origin: account, + Target: account, }) } } // Process accreted messages in serial. for _, msg := range msgs { - if err := p.state.Workers.ProcessFromClientAPI(ctx, msg); err != nil { + if err := p.state.Workers.Client.Process(ctx, msg); err != nil { log.Errorf( ctx, "error processing %s of %s during Delete of account %s: %v", diff --git a/internal/processing/account/follow.go b/internal/processing/account/follow.go index 7f28bb2c0..402c764e8 100644 --- a/internal/processing/account/follow.go +++ b/internal/processing/account/follow.go @@ -117,12 +117,12 @@ func (p *Processor) FollowCreate(ctx context.Context, requestingAccount *gtsmode } else { // Otherwise we leave the follow request as it is, // and we handle the rest of the process async. - p.state.Workers.EnqueueClientAPI(ctx, messages.FromClientAPI{ + p.state.Workers.Client.Queue.Push(&messages.FromClientAPI{ APObjectType: ap.ActivityFollow, APActivityType: ap.ActivityCreate, GTSModel: fr, - OriginAccount: requestingAccount, - TargetAccount: targetAccount, + Origin: requestingAccount, + Target: targetAccount, }) } @@ -143,7 +143,7 @@ func (p *Processor) FollowRemove(ctx context.Context, requestingAccount *gtsmode } // Batch queue accreted client api messages. - p.state.Workers.EnqueueClientAPI(ctx, msgs...) + p.state.Workers.Client.Queue.Push(msgs...) return p.RelationshipGet(ctx, requestingAccount, targetAccountID) } @@ -225,8 +225,8 @@ func (p *Processor) getFollowTarget(ctx context.Context, requester *gtsmodel.Acc // If a follow and/or follow request was removed this way, one or two // messages will be returned which should then be processed by a client // api worker. -func (p *Processor) unfollow(ctx context.Context, requestingAccount *gtsmodel.Account, targetAccount *gtsmodel.Account) ([]messages.FromClientAPI, error) { - var msgs []messages.FromClientAPI +func (p *Processor) unfollow(ctx context.Context, requestingAccount *gtsmodel.Account, targetAccount *gtsmodel.Account) ([]*messages.FromClientAPI, error) { + var msgs []*messages.FromClientAPI // Get follow from requesting account to target account. follow, err := p.state.DB.GetFollow(ctx, requestingAccount.ID, targetAccount.ID) @@ -251,7 +251,7 @@ func (p *Processor) unfollow(ctx context.Context, requestingAccount *gtsmodel.Ac } // Follow status changed, process side effects. - msgs = append(msgs, messages.FromClientAPI{ + msgs = append(msgs, &messages.FromClientAPI{ APObjectType: ap.ActivityFollow, APActivityType: ap.ActivityUndo, GTSModel: >smodel.Follow{ @@ -259,8 +259,8 @@ func (p *Processor) unfollow(ctx context.Context, requestingAccount *gtsmodel.Ac TargetAccountID: targetAccount.ID, URI: follow.URI, }, - OriginAccount: requestingAccount, - TargetAccount: targetAccount, + Origin: requestingAccount, + Target: targetAccount, }) } @@ -287,7 +287,7 @@ func (p *Processor) unfollow(ctx context.Context, requestingAccount *gtsmodel.Ac } // Follow status changed, process side effects. - msgs = append(msgs, messages.FromClientAPI{ + msgs = append(msgs, &messages.FromClientAPI{ APObjectType: ap.ActivityFollow, APActivityType: ap.ActivityUndo, GTSModel: >smodel.Follow{ @@ -295,8 +295,8 @@ func (p *Processor) unfollow(ctx context.Context, requestingAccount *gtsmodel.Ac TargetAccountID: targetAccount.ID, URI: followReq.URI, }, - OriginAccount: requestingAccount, - TargetAccount: targetAccount, + Origin: requestingAccount, + Target: targetAccount, }) } diff --git a/internal/processing/account/follow_request.go b/internal/processing/account/follow_request.go index c054637c8..6f6c7ba2d 100644 --- a/internal/processing/account/follow_request.go +++ b/internal/processing/account/follow_request.go @@ -40,12 +40,12 @@ func (p *Processor) FollowRequestAccept(ctx context.Context, requestingAccount * if follow.Account != nil { // Only enqueue work in the case we have a request creating account stored. // NOTE: due to how AcceptFollowRequest works, the inverse shouldn't be possible. - p.state.Workers.EnqueueClientAPI(ctx, messages.FromClientAPI{ + p.state.Workers.Client.Queue.Push(&messages.FromClientAPI{ APObjectType: ap.ActivityFollow, APActivityType: ap.ActivityAccept, GTSModel: follow, - OriginAccount: follow.Account, - TargetAccount: follow.TargetAccount, + Origin: follow.Account, + Target: follow.TargetAccount, }) } @@ -67,12 +67,12 @@ func (p *Processor) FollowRequestReject(ctx context.Context, requestingAccount * if followRequest.Account != nil { // Only enqueue work in the case we have a request creating account stored. // NOTE: due to how GetFollowRequest works, the inverse shouldn't be possible. - p.state.Workers.EnqueueClientAPI(ctx, messages.FromClientAPI{ + p.state.Workers.Client.Queue.Push(&messages.FromClientAPI{ APObjectType: ap.ActivityFollow, APActivityType: ap.ActivityReject, GTSModel: followRequest, - OriginAccount: followRequest.Account, - TargetAccount: followRequest.TargetAccount, + Origin: followRequest.Account, + Target: followRequest.TargetAccount, }) } diff --git a/internal/processing/account/follow_test.go b/internal/processing/account/follow_test.go index c269dc710..9ea8ce1b8 100644 --- a/internal/processing/account/follow_test.go +++ b/internal/processing/account/follow_test.go @@ -25,7 +25,6 @@ import ( "github.com/stretchr/testify/suite" "github.com/superseriousbusiness/gotosocial/internal/ap" apimodel "github.com/superseriousbusiness/gotosocial/internal/api/model" - "github.com/superseriousbusiness/gotosocial/internal/messages" "github.com/superseriousbusiness/gotosocial/internal/util" ) @@ -152,18 +151,11 @@ func (suite *FollowTestSuite) TestFollowRequestLocal() { } // There should be a message going to the worker. - var cMsg messages.FromClientAPI - select { - case cMsg = <-suite.fromClientAPIChan: - // No problem. - case <-time.After(5 * time.Second): - suite.FailNow("timed out waiting for message") - } - + cMsg, _ := suite.getClientMsg(5 * time.Second) suite.Equal(ap.ActivityCreate, cMsg.APActivityType) suite.Equal(ap.ActivityFollow, cMsg.APObjectType) - suite.Equal(requestingAccount.ID, cMsg.OriginAccount.ID) - suite.Equal(targetAccount.ID, cMsg.TargetAccount.ID) + suite.Equal(requestingAccount.ID, cMsg.Origin.ID) + suite.Equal(targetAccount.ID, cMsg.Target.ID) } func TestFollowTestS(t *testing.T) { diff --git a/internal/processing/account/move.go b/internal/processing/account/move.go index 602e8c021..63187dfd1 100644 --- a/internal/processing/account/move.go +++ b/internal/processing/account/move.go @@ -317,12 +317,12 @@ func (p *Processor) MoveSelf( } // Everything seems OK, process Move side effects async. - p.state.Workers.EnqueueClientAPI(ctx, messages.FromClientAPI{ + p.state.Workers.Client.Queue.Push(&messages.FromClientAPI{ APObjectType: ap.ActorPerson, APActivityType: ap.ActivityMove, GTSModel: move, - OriginAccount: originAcct, - TargetAccount: targetAcct, + Origin: originAcct, + Target: targetAcct, }) return nil diff --git a/internal/processing/account/move_test.go b/internal/processing/account/move_test.go index dfa0ea4e4..c1a931252 100644 --- a/internal/processing/account/move_test.go +++ b/internal/processing/account/move_test.go @@ -70,29 +70,23 @@ func (suite *MoveTestSuite) TestMoveAccountOK() { suite.FailNow(err.Error()) } - // There should be a msg heading back to fromClientAPI. - select { - case msg := <-suite.fromClientAPIChan: - move, ok := msg.GTSModel.(*gtsmodel.Move) - if !ok { - suite.FailNow("", "could not cast %T to *gtsmodel.Move", move) - } - - now := time.Now() - suite.WithinDuration(now, move.CreatedAt, 5*time.Second) - suite.WithinDuration(now, move.UpdatedAt, 5*time.Second) - suite.WithinDuration(now, move.AttemptedAt, 5*time.Second) - suite.Zero(move.SucceededAt) - suite.NotZero(move.ID) - suite.Equal(requestingAcct.URI, move.OriginURI) - suite.NotNil(move.Origin) - suite.Equal(targetAcct.URI, move.TargetURI) - suite.NotNil(move.Target) - suite.NotZero(move.URI) - - case <-time.After(5 * time.Second): - suite.FailNow("time out waiting for message") + // There should be a message going to the worker. + cMsg, _ := suite.getClientMsg(5 * time.Second) + move, ok := cMsg.GTSModel.(*gtsmodel.Move) + if !ok { + suite.FailNow("", "could not cast %T to *gtsmodel.Move", move) } + now := time.Now() + suite.WithinDuration(now, move.CreatedAt, 5*time.Second) + suite.WithinDuration(now, move.UpdatedAt, 5*time.Second) + suite.WithinDuration(now, move.AttemptedAt, 5*time.Second) + suite.Zero(move.SucceededAt) + suite.NotZero(move.ID) + suite.Equal(requestingAcct.URI, move.OriginURI) + suite.NotNil(move.Origin) + suite.Equal(targetAcct.URI, move.TargetURI) + suite.NotNil(move.Target) + suite.NotZero(move.URI) // Move should be in the database now. move, err := suite.state.DB.GetMoveByOriginTarget( diff --git a/internal/processing/account/update.go b/internal/processing/account/update.go index 670620e19..7f2749503 100644 --- a/internal/processing/account/update.go +++ b/internal/processing/account/update.go @@ -296,11 +296,11 @@ func (p *Processor) Update(ctx context.Context, account *gtsmodel.Account, form return nil, gtserror.NewErrorInternalError(fmt.Errorf("could not update account settings %s: %s", account.ID, err)) } - p.state.Workers.EnqueueClientAPI(ctx, messages.FromClientAPI{ + p.state.Workers.Client.Queue.Push(&messages.FromClientAPI{ APObjectType: ap.ObjectProfile, APActivityType: ap.ActivityUpdate, GTSModel: account, - OriginAccount: account, + Origin: account, }) acctSensitive, err := p.converter.AccountToAPIAccountSensitive(ctx, account) diff --git a/internal/processing/account/update_test.go b/internal/processing/account/update_test.go index 76ad3abe8..ad09ff25c 100644 --- a/internal/processing/account/update_test.go +++ b/internal/processing/account/update_test.go @@ -20,6 +20,7 @@ package account_test import ( "context" "testing" + "time" "github.com/stretchr/testify/suite" "github.com/superseriousbusiness/gotosocial/internal/ap" @@ -31,20 +32,6 @@ type AccountUpdateTestSuite struct { AccountStandardTestSuite } -func (suite *AccountStandardTestSuite) checkClientAPIChan(accountID string) { - msg := <-suite.fromClientAPIChan - - // Profile update. - suite.Equal(ap.ActivityUpdate, msg.APActivityType) - suite.Equal(ap.ObjectProfile, msg.APObjectType) - - // Correct account updated. - if msg.OriginAccount == nil { - suite.FailNow("expected msg.OriginAccount not to be nil") - } - suite.Equal(accountID, msg.OriginAccount.ID) -} - func (suite *AccountUpdateTestSuite) TestAccountUpdateSimple() { testAccount := >smodel.Account{} *testAccount = *suite.testAccounts["local_account_1"] @@ -73,7 +60,17 @@ func (suite *AccountUpdateTestSuite) TestAccountUpdateSimple() { suite.Equal(noteExpected, apiAccount.Note) // We should have an update in the client api channel. - suite.checkClientAPIChan(testAccount.ID) + msg, _ := suite.getClientMsg(5 * time.Second) + + // Profile update. + suite.Equal(ap.ActivityUpdate, msg.APActivityType) + suite.Equal(ap.ObjectProfile, msg.APObjectType) + + // Correct account updated. + if msg.Origin == nil { + suite.FailNow("expected msg.OriginAccount not to be nil") + } + suite.Equal(testAccount.ID, msg.Origin.ID) // Check database model of account as well. dbAccount, err := suite.db.GetAccountByID(ctx, testAccount.ID) @@ -113,7 +110,17 @@ func (suite *AccountUpdateTestSuite) TestAccountUpdateWithMention() { suite.Equal(noteExpected, apiAccount.Note) // We should have an update in the client api channel. - suite.checkClientAPIChan(testAccount.ID) + msg, _ := suite.getClientMsg(5 * time.Second) + + // Profile update. + suite.Equal(ap.ActivityUpdate, msg.APActivityType) + suite.Equal(ap.ObjectProfile, msg.APObjectType) + + // Correct account updated. + if msg.Origin == nil { + suite.FailNow("expected msg.OriginAccount not to be nil") + } + suite.Equal(testAccount.ID, msg.Origin.ID) // Check database model of account as well. dbAccount, err := suite.db.GetAccountByID(ctx, testAccount.ID) @@ -159,7 +166,17 @@ func (suite *AccountUpdateTestSuite) TestAccountUpdateWithMarkdownNote() { suite.Equal(noteExpected, apiAccount.Note) // We should have an update in the client api channel. - suite.checkClientAPIChan(testAccount.ID) + msg, _ := suite.getClientMsg(5 * time.Second) + + // Profile update. + suite.Equal(ap.ActivityUpdate, msg.APActivityType) + suite.Equal(ap.ObjectProfile, msg.APObjectType) + + // Correct account updated. + if msg.Origin == nil { + suite.FailNow("expected msg.OriginAccount not to be nil") + } + suite.Equal(testAccount.ID, msg.Origin.ID) // Check database model of account as well. dbAccount, err := suite.db.GetAccountByID(ctx, testAccount.ID) @@ -234,7 +251,17 @@ func (suite *AccountUpdateTestSuite) TestAccountUpdateWithFields() { suite.EqualValues(emojisExpected, apiAccount.Emojis) // We should have an update in the client api channel. - suite.checkClientAPIChan(testAccount.ID) + msg, _ := suite.getClientMsg(5 * time.Second) + + // Profile update. + suite.Equal(ap.ActivityUpdate, msg.APActivityType) + suite.Equal(ap.ObjectProfile, msg.APObjectType) + + // Correct account updated. + if msg.Origin == nil { + suite.FailNow("expected msg.OriginAccount not to be nil") + } + suite.Equal(testAccount.ID, msg.Origin.ID) // Check database model of account as well. dbAccount, err := suite.db.GetAccountByID(ctx, testAccount.ID) @@ -281,7 +308,17 @@ func (suite *AccountUpdateTestSuite) TestAccountUpdateNoteNotFields() { suite.Equal(fieldsBefore, len(apiAccount.Fields)) // We should have an update in the client api channel. - suite.checkClientAPIChan(testAccount.ID) + msg, _ := suite.getClientMsg(5 * time.Second) + + // Profile update. + suite.Equal(ap.ActivityUpdate, msg.APActivityType) + suite.Equal(ap.ObjectProfile, msg.APObjectType) + + // Correct account updated. + if msg.Origin == nil { + suite.FailNow("expected msg.OriginAccount not to be nil") + } + suite.Equal(testAccount.ID, msg.Origin.ID) // Check database model of account as well. dbAccount, err := suite.db.GetAccountByID(ctx, testAccount.ID) 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) diff --git a/internal/processing/polls/expiry.go b/internal/processing/polls/expiry.go index 59d0f17fe..d02a05f0d 100644 --- a/internal/processing/polls/expiry.go +++ b/internal/processing/polls/expiry.go @@ -116,11 +116,11 @@ func (p *Processor) onExpiry(pollID string) func(context.Context, time.Time) { // Enqueue a status update operation to the client API worker, // this will asynchronously send an update with the Poll close time. - p.state.Workers.EnqueueClientAPI(ctx, messages.FromClientAPI{ + p.state.Workers.Client.Queue.Push(&messages.FromClientAPI{ APActivityType: ap.ActivityUpdate, APObjectType: ap.ObjectNote, GTSModel: status, - OriginAccount: status.Account, + Origin: status.Account, }) } } diff --git a/internal/processing/polls/vote.go b/internal/processing/polls/vote.go index 5bc523978..c970fe106 100644 --- a/internal/processing/polls/vote.go +++ b/internal/processing/polls/vote.go @@ -96,11 +96,11 @@ func (p *Processor) PollVote(ctx context.Context, requester *gtsmodel.Account, p poll.IncrementVotes(choices) // Enqueue worker task to handle side-effects of user poll vote(s). - p.state.Workers.EnqueueClientAPI(ctx, messages.FromClientAPI{ + p.state.Workers.Client.Queue.Push(&messages.FromClientAPI{ APActivityType: ap.ActivityCreate, APObjectType: ap.ActivityQuestion, GTSModel: vote, // the vote choices - OriginAccount: requester, + Origin: requester, }) // Return converted API model poll. diff --git a/internal/processing/processor_test.go b/internal/processing/processor_test.go index 2a1d47e46..767e8b5ef 100644 --- a/internal/processing/processor_test.go +++ b/internal/processing/processor_test.go @@ -95,7 +95,6 @@ func (suite *ProcessingStandardTestSuite) SetupSuite() { func (suite *ProcessingStandardTestSuite) SetupTest() { suite.state.Caches.Init() - testrig.StartNoopWorkers(&suite.state) testrig.InitTestConfig() testrig.InitTestLog() @@ -124,8 +123,7 @@ func (suite *ProcessingStandardTestSuite) SetupTest() { suite.emailSender = testrig.NewEmailSender("../../web/template/", nil) suite.processor = processing.NewProcessor(cleaner.New(&suite.state), suite.typeconverter, suite.federator, suite.oauthServer, suite.mediaManager, &suite.state, suite.emailSender) - suite.state.Workers.EnqueueClientAPI = suite.processor.Workers().EnqueueClientAPI - suite.state.Workers.EnqueueFediAPI = suite.processor.Workers().EnqueueFediAPI + testrig.StartWorkers(&suite.state, suite.processor.Workers()) testrig.StandardDBSetup(suite.db, suite.testAccounts) testrig.StandardStorageSetup(suite.storage, "../../testrig/media") diff --git a/internal/processing/report/create.go b/internal/processing/report/create.go index c65ae0d52..cac600006 100644 --- a/internal/processing/report/create.go +++ b/internal/processing/report/create.go @@ -91,12 +91,12 @@ func (p *Processor) Create(ctx context.Context, account *gtsmodel.Account, form return nil, gtserror.NewErrorInternalError(err) } - p.state.Workers.EnqueueClientAPI(ctx, messages.FromClientAPI{ + p.state.Workers.Client.Queue.Push(&messages.FromClientAPI{ APObjectType: ap.ObjectProfile, APActivityType: ap.ActivityFlag, GTSModel: report, - OriginAccount: account, - TargetAccount: targetAccount, + Origin: account, + Target: targetAccount, }) apiReport, err := p.converter.ReportToAPIReport(ctx, report) diff --git a/internal/processing/status/boost.go b/internal/processing/status/boost.go index 5fae695fd..1c1da4ca7 100644 --- a/internal/processing/status/boost.go +++ b/internal/processing/status/boost.go @@ -89,12 +89,12 @@ func (p *Processor) BoostCreate( } // Process side effects asynchronously. - p.state.Workers.EnqueueClientAPI(ctx, messages.FromClientAPI{ + p.state.Workers.Client.Queue.Push(&messages.FromClientAPI{ APObjectType: ap.ActivityAnnounce, APActivityType: ap.ActivityCreate, GTSModel: boost, - OriginAccount: requester, - TargetAccount: target.Account, + Origin: requester, + Target: target.Account, }) return p.c.GetAPIStatus(ctx, requester, boost) @@ -141,12 +141,12 @@ func (p *Processor) BoostRemove( if boost != nil { // Status was boosted. Process unboost side effects asynchronously. - p.state.Workers.EnqueueClientAPI(ctx, messages.FromClientAPI{ + p.state.Workers.Client.Queue.Push(&messages.FromClientAPI{ APObjectType: ap.ActivityAnnounce, APActivityType: ap.ActivityUndo, GTSModel: boost, - OriginAccount: requester, - TargetAccount: target.Account, + Origin: requester, + Target: target.Account, }) } diff --git a/internal/processing/status/create.go b/internal/processing/status/create.go index d758fc0fb..80cc65c7f 100644 --- a/internal/processing/status/create.go +++ b/internal/processing/status/create.go @@ -143,11 +143,11 @@ func (p *Processor) Create( } // send it back to the client API worker for async side-effects. - p.state.Workers.EnqueueClientAPI(ctx, messages.FromClientAPI{ + p.state.Workers.Client.Queue.Push(&messages.FromClientAPI{ APObjectType: ap.ObjectNote, APActivityType: ap.ActivityCreate, GTSModel: status, - OriginAccount: requester, + Origin: requester, }) if status.Poll != nil { diff --git a/internal/processing/status/delete.go b/internal/processing/status/delete.go index 261086bdb..700909f44 100644 --- a/internal/processing/status/delete.go +++ b/internal/processing/status/delete.go @@ -51,12 +51,12 @@ func (p *Processor) Delete(ctx context.Context, requestingAccount *gtsmodel.Acco } // Process delete side effects. - p.state.Workers.EnqueueClientAPI(ctx, messages.FromClientAPI{ + p.state.Workers.Client.Queue.Push(&messages.FromClientAPI{ APObjectType: ap.ObjectNote, APActivityType: ap.ActivityDelete, GTSModel: targetStatus, - OriginAccount: requestingAccount, - TargetAccount: requestingAccount, + Origin: requestingAccount, + Target: requestingAccount, }) return apiStatus, nil diff --git a/internal/processing/status/fave.go b/internal/processing/status/fave.go index 7b71725ab..dd961c082 100644 --- a/internal/processing/status/fave.go +++ b/internal/processing/status/fave.go @@ -107,12 +107,12 @@ func (p *Processor) FaveCreate(ctx context.Context, requestingAccount *gtsmodel. } // Process new status fave side effects. - p.state.Workers.EnqueueClientAPI(ctx, messages.FromClientAPI{ + p.state.Workers.Client.Queue.Push(&messages.FromClientAPI{ APObjectType: ap.ActivityLike, APActivityType: ap.ActivityCreate, GTSModel: gtsFave, - OriginAccount: requestingAccount, - TargetAccount: targetStatus.Account, + Origin: requestingAccount, + Target: targetStatus.Account, }) return p.c.GetAPIStatus(ctx, requestingAccount, targetStatus) @@ -137,12 +137,12 @@ func (p *Processor) FaveRemove(ctx context.Context, requestingAccount *gtsmodel. } // Process remove status fave side effects. - p.state.Workers.EnqueueClientAPI(ctx, messages.FromClientAPI{ + p.state.Workers.Client.Queue.Push(&messages.FromClientAPI{ APObjectType: ap.ActivityLike, APActivityType: ap.ActivityUndo, GTSModel: existingFave, - OriginAccount: requestingAccount, - TargetAccount: targetStatus.Account, + Origin: requestingAccount, + Target: targetStatus.Account, }) return p.c.GetAPIStatus(ctx, requestingAccount, targetStatus) diff --git a/internal/processing/workers/federate.go b/internal/processing/workers/federate.go index e737513f5..3538c9958 100644 --- a/internal/processing/workers/federate.go +++ b/internal/processing/workers/federate.go @@ -75,12 +75,6 @@ func (f *federate) DeleteAccount(ctx context.Context, account *gtsmodel.Account) return nil } - // Drop any queued outgoing AP requests to / from account, - // (this stops any 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) - // Parse relevant URI(s). outboxIRI, err := parseURI(account.OutboxURI) if err != nil { @@ -228,16 +222,6 @@ func (f *federate) DeleteStatus(ctx context.Context, status *gtsmodel.Status) er return nil } - // Drop any queued outgoing http requests for status, - // (this stops any queued likes, boosts, creates etc). - f.state.Workers.Delivery.Queue.Delete("ObjectID", status.URI) - f.state.Workers.Delivery.Queue.Delete("TargetID", 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) - } - // Parse the outbox URI of the status author. outboxIRI, err := parseURI(status.Account.OutboxURI) if err != nil { diff --git a/internal/processing/workers/fromclientapi.go b/internal/processing/workers/fromclientapi.go index 1412ea003..4564afbb9 100644 --- a/internal/processing/workers/fromclientapi.go +++ b/internal/processing/workers/fromclientapi.go @@ -25,7 +25,6 @@ import ( "codeberg.org/gruf/go-logger/v2/level" "github.com/superseriousbusiness/gotosocial/internal/ap" "github.com/superseriousbusiness/gotosocial/internal/db" - "github.com/superseriousbusiness/gotosocial/internal/gtscontext" "github.com/superseriousbusiness/gotosocial/internal/gtserror" "github.com/superseriousbusiness/gotosocial/internal/gtsmodel" "github.com/superseriousbusiness/gotosocial/internal/log" @@ -45,29 +44,15 @@ type clientAPI struct { surface *surface federate *federate account *account.Processor - utilF *utilF + utils *utils } -func (p *Processor) EnqueueClientAPI(cctx context.Context, msgs ...messages.FromClientAPI) { - _ = p.workers.ClientAPI.MustEnqueueCtx(cctx, func(wctx context.Context) { - // Copy caller ctx values to worker's. - wctx = gtscontext.WithValues(wctx, cctx) - - // Process worker messages. - for _, msg := range msgs { - if err := p.ProcessFromClientAPI(wctx, msg); err != nil { - log.Errorf(wctx, "error processing client API message: %v", err) - } - } - }) -} - -func (p *Processor) ProcessFromClientAPI(ctx context.Context, cMsg messages.FromClientAPI) error { +func (p *Processor) ProcessFromClientAPI(ctx context.Context, cMsg *messages.FromClientAPI) error { // Allocate new log fields slice fields := make([]kv.Field, 3, 4) fields[0] = kv.Field{"activityType", cMsg.APActivityType} fields[1] = kv.Field{"objectType", cMsg.APObjectType} - fields[2] = kv.Field{"fromAccount", cMsg.OriginAccount.Username} + fields[2] = kv.Field{"fromAccount", cMsg.Origin.Username} // Include GTSModel in logs if appropriate. if cMsg.GTSModel != nil && @@ -217,7 +202,7 @@ func (p *Processor) ProcessFromClientAPI(ctx context.Context, cMsg messages.From return gtserror.Newf("unhandled: %s %s", cMsg.APActivityType, cMsg.APObjectType) } -func (p *clientAPI) CreateAccount(ctx context.Context, cMsg messages.FromClientAPI) error { +func (p *clientAPI) CreateAccount(ctx context.Context, cMsg *messages.FromClientAPI) error { newUser, ok := cMsg.GTSModel.(*gtsmodel.User) if !ok { return gtserror.Newf("%T not parseable as *gtsmodel.User", cMsg.GTSModel) @@ -241,14 +226,14 @@ func (p *clientAPI) CreateAccount(ctx context.Context, cMsg messages.FromClientA return nil } -func (p *clientAPI) CreateStatus(ctx context.Context, cMsg messages.FromClientAPI) error { +func (p *clientAPI) CreateStatus(ctx context.Context, cMsg *messages.FromClientAPI) error { status, ok := cMsg.GTSModel.(*gtsmodel.Status) if !ok { return gtserror.Newf("%T not parseable as *gtsmodel.Status", cMsg.GTSModel) } // Update stats for the actor account. - if err := p.utilF.incrementStatusesCount(ctx, cMsg.OriginAccount, status); err != nil { + if err := p.utils.incrementStatusesCount(ctx, cMsg.Origin, status); err != nil { log.Errorf(ctx, "error updating account stats: %v", err) } @@ -269,7 +254,7 @@ func (p *clientAPI) CreateStatus(ctx context.Context, cMsg messages.FromClientAP return nil } -func (p *clientAPI) CreatePollVote(ctx context.Context, cMsg messages.FromClientAPI) error { +func (p *clientAPI) CreatePollVote(ctx context.Context, cMsg *messages.FromClientAPI) error { // Cast the create poll vote attached to message. vote, ok := cMsg.GTSModel.(*gtsmodel.PollVote) if !ok { @@ -310,14 +295,14 @@ func (p *clientAPI) CreatePollVote(ctx context.Context, cMsg messages.FromClient return nil } -func (p *clientAPI) CreateFollowReq(ctx context.Context, cMsg messages.FromClientAPI) error { +func (p *clientAPI) CreateFollowReq(ctx context.Context, cMsg *messages.FromClientAPI) error { followRequest, ok := cMsg.GTSModel.(*gtsmodel.FollowRequest) if !ok { return gtserror.Newf("%T not parseable as *gtsmodel.FollowRequest", cMsg.GTSModel) } // Update stats for the target account. - if err := p.utilF.incrementFollowRequestsCount(ctx, cMsg.TargetAccount); err != nil { + if err := p.utils.incrementFollowRequestsCount(ctx, cMsg.Target); err != nil { log.Errorf(ctx, "error updating account stats: %v", err) } @@ -338,7 +323,7 @@ func (p *clientAPI) CreateFollowReq(ctx context.Context, cMsg messages.FromClien return nil } -func (p *clientAPI) CreateLike(ctx context.Context, cMsg messages.FromClientAPI) error { +func (p *clientAPI) CreateLike(ctx context.Context, cMsg *messages.FromClientAPI) error { fave, ok := cMsg.GTSModel.(*gtsmodel.StatusFave) if !ok { return gtserror.Newf("%T not parseable as *gtsmodel.StatusFave", cMsg.GTSModel) @@ -364,14 +349,14 @@ func (p *clientAPI) CreateLike(ctx context.Context, cMsg messages.FromClientAPI) return nil } -func (p *clientAPI) CreateAnnounce(ctx context.Context, cMsg messages.FromClientAPI) error { +func (p *clientAPI) CreateAnnounce(ctx context.Context, cMsg *messages.FromClientAPI) error { boost, ok := cMsg.GTSModel.(*gtsmodel.Status) if !ok { return gtserror.Newf("%T not parseable as *gtsmodel.Status", cMsg.GTSModel) } // Update stats for the actor account. - if err := p.utilF.incrementStatusesCount(ctx, cMsg.OriginAccount, boost); err != nil { + if err := p.utils.incrementStatusesCount(ctx, cMsg.Origin, boost); err != nil { log.Errorf(ctx, "error updating account stats: %v", err) } @@ -396,7 +381,7 @@ func (p *clientAPI) CreateAnnounce(ctx context.Context, cMsg messages.FromClient return nil } -func (p *clientAPI) CreateBlock(ctx context.Context, cMsg messages.FromClientAPI) error { +func (p *clientAPI) CreateBlock(ctx context.Context, cMsg *messages.FromClientAPI) error { block, ok := cMsg.GTSModel.(*gtsmodel.Block) if !ok { return gtserror.Newf("%T not parseable as *gtsmodel.Block", cMsg.GTSModel) @@ -430,7 +415,7 @@ func (p *clientAPI) CreateBlock(ctx context.Context, cMsg messages.FromClientAPI return nil } -func (p *clientAPI) UpdateStatus(ctx context.Context, cMsg messages.FromClientAPI) error { +func (p *clientAPI) UpdateStatus(ctx context.Context, cMsg *messages.FromClientAPI) error { // Cast the updated Status model attached to msg. status, ok := cMsg.GTSModel.(*gtsmodel.Status) if !ok { @@ -462,7 +447,7 @@ func (p *clientAPI) UpdateStatus(ctx context.Context, cMsg messages.FromClientAP return nil } -func (p *clientAPI) UpdateAccount(ctx context.Context, cMsg messages.FromClientAPI) error { +func (p *clientAPI) UpdateAccount(ctx context.Context, cMsg *messages.FromClientAPI) error { account, ok := cMsg.GTSModel.(*gtsmodel.Account) if !ok { return gtserror.Newf("cannot cast %T -> *gtsmodel.Account", cMsg.GTSModel) @@ -475,7 +460,7 @@ func (p *clientAPI) UpdateAccount(ctx context.Context, cMsg messages.FromClientA return nil } -func (p *clientAPI) UpdateReport(ctx context.Context, cMsg messages.FromClientAPI) error { +func (p *clientAPI) UpdateReport(ctx context.Context, cMsg *messages.FromClientAPI) error { report, ok := cMsg.GTSModel.(*gtsmodel.Report) if !ok { return gtserror.Newf("%T not parseable as *gtsmodel.Report", cMsg.GTSModel) @@ -494,23 +479,23 @@ func (p *clientAPI) UpdateReport(ctx context.Context, cMsg messages.FromClientAP return nil } -func (p *clientAPI) AcceptFollow(ctx context.Context, cMsg messages.FromClientAPI) error { +func (p *clientAPI) AcceptFollow(ctx context.Context, cMsg *messages.FromClientAPI) error { follow, ok := cMsg.GTSModel.(*gtsmodel.Follow) if !ok { return gtserror.Newf("%T not parseable as *gtsmodel.Follow", cMsg.GTSModel) } // Update stats for the target account. - if err := p.utilF.decrementFollowRequestsCount(ctx, cMsg.TargetAccount); err != nil { + if err := p.utils.decrementFollowRequestsCount(ctx, cMsg.Target); err != nil { log.Errorf(ctx, "error updating account stats: %v", err) } - if err := p.utilF.incrementFollowersCount(ctx, cMsg.TargetAccount); err != nil { + if err := p.utils.incrementFollowersCount(ctx, cMsg.Target); err != nil { log.Errorf(ctx, "error updating account stats: %v", err) } // Update stats for the origin account. - if err := p.utilF.incrementFollowingCount(ctx, cMsg.OriginAccount); err != nil { + if err := p.utils.incrementFollowingCount(ctx, cMsg.Origin); err != nil { log.Errorf(ctx, "error updating account stats: %v", err) } @@ -525,14 +510,14 @@ func (p *clientAPI) AcceptFollow(ctx context.Context, cMsg messages.FromClientAP return nil } -func (p *clientAPI) RejectFollowRequest(ctx context.Context, cMsg messages.FromClientAPI) error { +func (p *clientAPI) RejectFollowRequest(ctx context.Context, cMsg *messages.FromClientAPI) error { followReq, ok := cMsg.GTSModel.(*gtsmodel.FollowRequest) if !ok { return gtserror.Newf("%T not parseable as *gtsmodel.FollowRequest", cMsg.GTSModel) } // Update stats for the target account. - if err := p.utilF.decrementFollowRequestsCount(ctx, cMsg.TargetAccount); err != nil { + if err := p.utils.decrementFollowRequestsCount(ctx, cMsg.Target); err != nil { log.Errorf(ctx, "error updating account stats: %v", err) } @@ -546,19 +531,19 @@ func (p *clientAPI) RejectFollowRequest(ctx context.Context, cMsg messages.FromC return nil } -func (p *clientAPI) UndoFollow(ctx context.Context, cMsg messages.FromClientAPI) error { +func (p *clientAPI) UndoFollow(ctx context.Context, cMsg *messages.FromClientAPI) error { follow, ok := cMsg.GTSModel.(*gtsmodel.Follow) if !ok { return gtserror.Newf("%T not parseable as *gtsmodel.Follow", cMsg.GTSModel) } // Update stats for the origin account. - if err := p.utilF.decrementFollowingCount(ctx, cMsg.OriginAccount); err != nil { + if err := p.utils.decrementFollowingCount(ctx, cMsg.Origin); err != nil { log.Errorf(ctx, "error updating account stats: %v", err) } // Update stats for the target account. - if err := p.utilF.decrementFollowersCount(ctx, cMsg.TargetAccount); err != nil { + if err := p.utils.decrementFollowersCount(ctx, cMsg.Target); err != nil { log.Errorf(ctx, "error updating account stats: %v", err) } @@ -569,7 +554,7 @@ func (p *clientAPI) UndoFollow(ctx context.Context, cMsg messages.FromClientAPI) return nil } -func (p *clientAPI) UndoBlock(ctx context.Context, cMsg messages.FromClientAPI) error { +func (p *clientAPI) UndoBlock(ctx context.Context, cMsg *messages.FromClientAPI) error { block, ok := cMsg.GTSModel.(*gtsmodel.Block) if !ok { return gtserror.Newf("%T not parseable as *gtsmodel.Block", cMsg.GTSModel) @@ -582,7 +567,7 @@ func (p *clientAPI) UndoBlock(ctx context.Context, cMsg messages.FromClientAPI) return nil } -func (p *clientAPI) UndoFave(ctx context.Context, cMsg messages.FromClientAPI) error { +func (p *clientAPI) UndoFave(ctx context.Context, cMsg *messages.FromClientAPI) error { statusFave, ok := cMsg.GTSModel.(*gtsmodel.StatusFave) if !ok { return gtserror.Newf("%T not parseable as *gtsmodel.StatusFave", cMsg.GTSModel) @@ -599,7 +584,7 @@ func (p *clientAPI) UndoFave(ctx context.Context, cMsg messages.FromClientAPI) e return nil } -func (p *clientAPI) UndoAnnounce(ctx context.Context, cMsg messages.FromClientAPI) error { +func (p *clientAPI) UndoAnnounce(ctx context.Context, cMsg *messages.FromClientAPI) error { status, ok := cMsg.GTSModel.(*gtsmodel.Status) if !ok { return gtserror.Newf("%T not parseable as *gtsmodel.Status", cMsg.GTSModel) @@ -610,7 +595,7 @@ func (p *clientAPI) UndoAnnounce(ctx context.Context, cMsg messages.FromClientAP } // Update stats for the origin account. - if err := p.utilF.decrementStatusesCount(ctx, cMsg.OriginAccount); err != nil { + if err := p.utils.decrementStatusesCount(ctx, cMsg.Origin); err != nil { log.Errorf(ctx, "error updating account stats: %v", err) } @@ -629,7 +614,7 @@ func (p *clientAPI) UndoAnnounce(ctx context.Context, cMsg messages.FromClientAP return nil } -func (p *clientAPI) DeleteStatus(ctx context.Context, cMsg messages.FromClientAPI) error { +func (p *clientAPI) DeleteStatus(ctx context.Context, cMsg *messages.FromClientAPI) error { // Don't delete attachments, just unattach them: // this request comes from the client API and the // poster may want to use attachments again later. @@ -648,12 +633,26 @@ func (p *clientAPI) DeleteStatus(ctx context.Context, cMsg messages.FromClientAP return gtserror.Newf("db error populating status: %w", err) } - if err := p.utilF.wipeStatus(ctx, status, deleteAttachments); err != nil { + // 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) + + // First perform the actual status deletion. + if err := p.utils.wipeStatus(ctx, status, deleteAttachments); err != nil { log.Errorf(ctx, "error wiping status: %v", err) } // Update stats for the origin account. - if err := p.utilF.decrementStatusesCount(ctx, cMsg.OriginAccount); err != nil { + if err := p.utils.decrementStatusesCount(ctx, cMsg.Origin); err != nil { log.Errorf(ctx, "error updating account stats: %v", err) } @@ -670,7 +669,7 @@ func (p *clientAPI) DeleteStatus(ctx context.Context, cMsg messages.FromClientAP return nil } -func (p *clientAPI) DeleteAccount(ctx context.Context, cMsg messages.FromClientAPI) error { +func (p *clientAPI) DeleteAccount(ctx context.Context, cMsg *messages.FromClientAPI) error { // The originID of the delete, one of: // - ID of a domain block, for which // this account delete is a side effect. @@ -684,21 +683,41 @@ func (p *clientAPI) DeleteAccount(ctx context.Context, cMsg messages.FromClientA } else { // Origin is whichever account // originated this message. - originID = cMsg.OriginAccount.ID + originID = cMsg.Origin.ID } - if err := p.federate.DeleteAccount(ctx, cMsg.TargetAccount); err != nil { + // Extract target account. + account := cMsg.Target + + // 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) } - if err := p.account.Delete(ctx, cMsg.TargetAccount, originID); err != nil { + if err := p.account.Delete(ctx, cMsg.Target, originID); err != nil { log.Errorf(ctx, "error deleting account: %v", err) } return nil } -func (p *clientAPI) ReportAccount(ctx context.Context, cMsg messages.FromClientAPI) error { +func (p *clientAPI) ReportAccount(ctx context.Context, cMsg *messages.FromClientAPI) error { report, ok := cMsg.GTSModel.(*gtsmodel.Report) if !ok { return gtserror.Newf("%T not parseable as *gtsmodel.Report", cMsg.GTSModel) @@ -719,28 +738,28 @@ func (p *clientAPI) ReportAccount(ctx context.Context, cMsg messages.FromClientA return nil } -func (p *clientAPI) MoveAccount(ctx context.Context, cMsg messages.FromClientAPI) error { +func (p *clientAPI) MoveAccount(ctx context.Context, cMsg *messages.FromClientAPI) error { // Redirect each local follower of // OriginAccount to follow move target. - p.utilF.redirectFollowers(ctx, cMsg.OriginAccount, cMsg.TargetAccount) + p.utils.redirectFollowers(ctx, cMsg.Origin, cMsg.Target) // At this point, we know OriginAccount has the // Move set on it. Just make sure it's populated. - if err := p.state.DB.PopulateMove(ctx, cMsg.OriginAccount.Move); err != nil { + if err := p.state.DB.PopulateMove(ctx, cMsg.Origin.Move); err != nil { return gtserror.Newf("error populating Move: %w", err) } // Now send the Move message out to // OriginAccount's (remote) followers. - if err := p.federate.MoveAccount(ctx, cMsg.OriginAccount); err != nil { + if err := p.federate.MoveAccount(ctx, cMsg.Origin); err != nil { return gtserror.Newf("error federating account move: %w", err) } // Mark the move attempt as successful. - cMsg.OriginAccount.Move.SucceededAt = cMsg.OriginAccount.Move.AttemptedAt + cMsg.Origin.Move.SucceededAt = cMsg.Origin.Move.AttemptedAt if err := p.state.DB.UpdateMove( ctx, - cMsg.OriginAccount.Move, + cMsg.Origin.Move, "succeeded_at", ); err != nil { return gtserror.Newf("error marking move as successful: %w", err) @@ -749,7 +768,7 @@ func (p *clientAPI) MoveAccount(ctx context.Context, cMsg messages.FromClientAPI return nil } -func (p *clientAPI) AcceptAccount(ctx context.Context, cMsg messages.FromClientAPI) error { +func (p *clientAPI) AcceptAccount(ctx context.Context, cMsg *messages.FromClientAPI) error { newUser, ok := cMsg.GTSModel.(*gtsmodel.User) if !ok { return gtserror.Newf("%T not parseable as *gtsmodel.User", cMsg.GTSModel) @@ -772,17 +791,17 @@ func (p *clientAPI) AcceptAccount(ctx context.Context, cMsg messages.FromClientA return nil } -func (p *clientAPI) RejectAccount(ctx context.Context, cMsg messages.FromClientAPI) error { +func (p *clientAPI) RejectAccount(ctx context.Context, cMsg *messages.FromClientAPI) error { deniedUser, ok := cMsg.GTSModel.(*gtsmodel.DeniedUser) if !ok { return gtserror.Newf("%T not parseable as *gtsmodel.DeniedUser", cMsg.GTSModel) } // Remove the account. - if err := p.state.DB.DeleteAccount(ctx, cMsg.TargetAccount.ID); err != nil { + if err := p.state.DB.DeleteAccount(ctx, cMsg.Target.ID); err != nil { log.Errorf(ctx, "db error deleting account %s: %v", - cMsg.TargetAccount.ID, err, + cMsg.Target.ID, err, ) } diff --git a/internal/processing/workers/fromclientapi_test.go b/internal/processing/workers/fromclientapi_test.go index 5e294597d..3aa0fbdad 100644 --- a/internal/processing/workers/fromclientapi_test.go +++ b/internal/processing/workers/fromclientapi_test.go @@ -197,11 +197,11 @@ func (suite *FromClientAPITestSuite) TestProcessCreateStatusWithNotification() { // Process the new status. if err := suite.processor.Workers().ProcessFromClientAPI( ctx, - messages.FromClientAPI{ + &messages.FromClientAPI{ APObjectType: ap.ObjectNote, APActivityType: ap.ActivityCreate, GTSModel: status, - OriginAccount: postingAccount, + Origin: postingAccount, }, ); err != nil { suite.FailNow(err.Error()) @@ -291,11 +291,11 @@ func (suite *FromClientAPITestSuite) TestProcessCreateStatusReply() { // Process the new status. if err := suite.processor.Workers().ProcessFromClientAPI( ctx, - messages.FromClientAPI{ + &messages.FromClientAPI{ APObjectType: ap.ObjectNote, APActivityType: ap.ActivityCreate, GTSModel: status, - OriginAccount: postingAccount, + Origin: postingAccount, }, ); err != nil { suite.FailNow(err.Error()) @@ -355,11 +355,11 @@ func (suite *FromClientAPITestSuite) TestProcessCreateStatusReplyMuted() { // Process the new status. if err := suite.processor.Workers().ProcessFromClientAPI( ctx, - messages.FromClientAPI{ + &messages.FromClientAPI{ APObjectType: ap.ObjectNote, APActivityType: ap.ActivityCreate, GTSModel: status, - OriginAccount: postingAccount, + Origin: postingAccount, }, ); err != nil { suite.FailNow(err.Error()) @@ -409,11 +409,11 @@ func (suite *FromClientAPITestSuite) TestProcessCreateStatusBoostMuted() { // Process the new status. if err := suite.processor.Workers().ProcessFromClientAPI( ctx, - messages.FromClientAPI{ + &messages.FromClientAPI{ APObjectType: ap.ActivityAnnounce, APActivityType: ap.ActivityCreate, GTSModel: status, - OriginAccount: postingAccount, + Origin: postingAccount, }, ); err != nil { suite.FailNow(err.Error()) @@ -467,11 +467,11 @@ func (suite *FromClientAPITestSuite) TestProcessCreateStatusListRepliesPolicyLis // Process the new status. if err := suite.processor.Workers().ProcessFromClientAPI( ctx, - messages.FromClientAPI{ + &messages.FromClientAPI{ APObjectType: ap.ObjectNote, APActivityType: ap.ActivityCreate, GTSModel: status, - OriginAccount: postingAccount, + Origin: postingAccount, }, ); err != nil { suite.FailNow(err.Error()) @@ -540,11 +540,11 @@ func (suite *FromClientAPITestSuite) TestProcessCreateStatusListRepliesPolicyLis // Process the new status. if err := suite.processor.Workers().ProcessFromClientAPI( ctx, - messages.FromClientAPI{ + &messages.FromClientAPI{ APObjectType: ap.ObjectNote, APActivityType: ap.ActivityCreate, GTSModel: status, - OriginAccount: postingAccount, + Origin: postingAccount, }, ); err != nil { suite.FailNow(err.Error()) @@ -608,11 +608,11 @@ func (suite *FromClientAPITestSuite) TestProcessCreateStatusReplyListRepliesPoli // Process the new status. if err := suite.processor.Workers().ProcessFromClientAPI( ctx, - messages.FromClientAPI{ + &messages.FromClientAPI{ APObjectType: ap.ObjectNote, APActivityType: ap.ActivityCreate, GTSModel: status, - OriginAccount: postingAccount, + Origin: postingAccount, }, ); err != nil { suite.FailNow(err.Error()) @@ -664,11 +664,11 @@ func (suite *FromClientAPITestSuite) TestProcessCreateStatusBoost() { // Process the new status. if err := suite.processor.Workers().ProcessFromClientAPI( ctx, - messages.FromClientAPI{ + &messages.FromClientAPI{ APObjectType: ap.ActivityAnnounce, APActivityType: ap.ActivityCreate, GTSModel: status, - OriginAccount: postingAccount, + Origin: postingAccount, }, ); err != nil { suite.FailNow(err.Error()) @@ -729,11 +729,11 @@ func (suite *FromClientAPITestSuite) TestProcessCreateStatusBoostNoReblogs() { // Process the new status. if err := suite.processor.Workers().ProcessFromClientAPI( ctx, - messages.FromClientAPI{ + &messages.FromClientAPI{ APObjectType: ap.ActivityAnnounce, APActivityType: ap.ActivityCreate, GTSModel: status, - OriginAccount: postingAccount, + Origin: postingAccount, }, ); err != nil { suite.FailNow(err.Error()) @@ -776,11 +776,11 @@ func (suite *FromClientAPITestSuite) TestProcessStatusDelete() { // Process the status delete. if err := suite.processor.Workers().ProcessFromClientAPI( ctx, - messages.FromClientAPI{ + &messages.FromClientAPI{ APObjectType: ap.ObjectNote, APActivityType: ap.ActivityDelete, GTSModel: deletedStatus, - OriginAccount: deletingAccount, + Origin: deletingAccount, }, ); err != nil { suite.FailNow(err.Error()) diff --git a/internal/processing/workers/fromfediapi.go b/internal/processing/workers/fromfediapi.go index 0b1106a9e..fcd5b38f2 100644 --- a/internal/processing/workers/fromfediapi.go +++ b/internal/processing/workers/fromfediapi.go @@ -19,13 +19,14 @@ 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/gtscontext" "github.com/superseriousbusiness/gotosocial/internal/gtserror" "github.com/superseriousbusiness/gotosocial/internal/gtsmodel" "github.com/superseriousbusiness/gotosocial/internal/log" @@ -43,34 +44,20 @@ type fediAPI struct { surface *surface federate *federate account *account.Processor - utilF *utilF + utils *utils } -func (p *Processor) EnqueueFediAPI(cctx context.Context, msgs ...messages.FromFediAPI) { - _ = p.workers.Federator.MustEnqueueCtx(cctx, func(wctx context.Context) { - // Copy caller ctx values to worker's. - wctx = gtscontext.WithValues(wctx, cctx) - - // Process worker messages. - for _, msg := range msgs { - if err := p.ProcessFromFediAPI(wctx, msg); err != nil { - log.Errorf(wctx, "error processing fedi API message: %v", err) - } - } - }) -} - -func (p *Processor) ProcessFromFediAPI(ctx context.Context, fMsg messages.FromFediAPI) error { +func (p *Processor) ProcessFromFediAPI(ctx context.Context, fMsg *messages.FromFediAPI) error { // Allocate new log fields slice fields := make([]kv.Field, 3, 5) fields[0] = kv.Field{"activityType", fMsg.APActivityType} fields[1] = kv.Field{"objectType", fMsg.APObjectType} - fields[2] = kv.Field{"toAccount", fMsg.ReceivingAccount.Username} + fields[2] = kv.Field{"toAccount", fMsg.Receiving.Username} - if fMsg.APIri != nil { + if fMsg.APIRI != nil { // An IRI was supplied, append to log fields = append(fields, kv.Field{ - "iri", fMsg.APIri, + "iri", fMsg.APIRI, }) } @@ -168,7 +155,7 @@ func (p *Processor) ProcessFromFediAPI(ctx context.Context, fMsg messages.FromFe return gtserror.Newf("unhandled: %s %s", fMsg.APActivityType, fMsg.APObjectType) } -func (p *fediAPI) CreateStatus(ctx context.Context, fMsg messages.FromFediAPI) error { +func (p *fediAPI) CreateStatus(ctx context.Context, fMsg *messages.FromFediAPI) error { var ( status *gtsmodel.Status statusable ap.Statusable @@ -178,11 +165,11 @@ func (p *fediAPI) CreateStatus(ctx context.Context, fMsg messages.FromFediAPI) e var ok bool switch { - case fMsg.APObjectModel != nil: + case fMsg.APObject != nil: // A model was provided, extract this from message. - statusable, ok = fMsg.APObjectModel.(ap.Statusable) + statusable, ok = fMsg.APObject.(ap.Statusable) if !ok { - return gtserror.Newf("cannot cast %T -> ap.Statusable", fMsg.APObjectModel) + return gtserror.Newf("cannot cast %T -> ap.Statusable", fMsg.APObject) } // Create bare-bones model to pass @@ -196,7 +183,7 @@ func (p *fediAPI) CreateStatus(ctx context.Context, fMsg messages.FromFediAPI) e // statusable model, which it will use to further flesh out // the bare bones model and insert it into the database. status, statusable, err = p.federate.RefreshStatus(ctx, - fMsg.ReceivingAccount.Username, + fMsg.Receiving.Username, bareStatus, statusable, // Force refresh within 5min window. @@ -206,15 +193,15 @@ func (p *fediAPI) CreateStatus(ctx context.Context, fMsg messages.FromFediAPI) e return gtserror.Newf("error processing new status %s: %w", bareStatus.URI, err) } - case fMsg.APIri != nil: + case fMsg.APIRI != nil: // Model was not set, deref with IRI (this is a forward). // This will also cause the status to be inserted into the db. status, statusable, err = p.federate.GetStatusByURI(ctx, - fMsg.ReceivingAccount.Username, - fMsg.APIri, + fMsg.Receiving.Username, + fMsg.APIRI, ) if err != nil { - return gtserror.Newf("error dereferencing forwarded status %s: %w", fMsg.APIri, err) + return gtserror.Newf("error dereferencing forwarded status %s: %w", fMsg.APIRI, err) } default: @@ -230,7 +217,7 @@ func (p *fediAPI) CreateStatus(ctx context.Context, fMsg messages.FromFediAPI) e } // Update stats for the remote account. - if err := p.utilF.incrementStatusesCount(ctx, fMsg.RequestingAccount, status); err != nil { + if err := p.utils.incrementStatusesCount(ctx, fMsg.Requesting, status); err != nil { log.Errorf(ctx, "error updating account stats: %v", err) } @@ -248,7 +235,7 @@ func (p *fediAPI) CreateStatus(ctx context.Context, fMsg messages.FromFediAPI) e return nil } -func (p *fediAPI) CreatePollVote(ctx context.Context, fMsg messages.FromFediAPI) error { +func (p *fediAPI) CreatePollVote(ctx context.Context, fMsg *messages.FromFediAPI) error { // Cast poll vote type from the worker message. vote, ok := fMsg.GTSModel.(*gtsmodel.PollVote) if !ok { @@ -293,7 +280,7 @@ func (p *fediAPI) CreatePollVote(ctx context.Context, fMsg messages.FromFediAPI) return nil } -func (p *fediAPI) CreateFollowReq(ctx context.Context, fMsg messages.FromFediAPI) error { +func (p *fediAPI) CreateFollowReq(ctx context.Context, fMsg *messages.FromFediAPI) error { followRequest, ok := fMsg.GTSModel.(*gtsmodel.FollowRequest) if !ok { return gtserror.Newf("%T not parseable as *gtsmodel.FollowRequest", fMsg.GTSModel) @@ -310,7 +297,7 @@ func (p *fediAPI) CreateFollowReq(ctx context.Context, fMsg messages.FromFediAPI } // And update stats for the local account. - if err := p.utilF.incrementFollowRequestsCount(ctx, fMsg.ReceivingAccount); err != nil { + if err := p.utils.incrementFollowRequestsCount(ctx, fMsg.Receiving); err != nil { log.Errorf(ctx, "error updating account stats: %v", err) } @@ -330,12 +317,12 @@ func (p *fediAPI) CreateFollowReq(ctx context.Context, fMsg messages.FromFediAPI } // Update stats for the local account. - if err := p.utilF.incrementFollowersCount(ctx, fMsg.ReceivingAccount); err != nil { + if err := p.utils.incrementFollowersCount(ctx, fMsg.Receiving); err != nil { log.Errorf(ctx, "error updating account stats: %v", err) } // Update stats for the remote account. - if err := p.utilF.incrementFollowingCount(ctx, fMsg.RequestingAccount); err != nil { + if err := p.utils.incrementFollowingCount(ctx, fMsg.Requesting); err != nil { log.Errorf(ctx, "error updating account stats: %v", err) } @@ -350,7 +337,7 @@ func (p *fediAPI) CreateFollowReq(ctx context.Context, fMsg messages.FromFediAPI return nil } -func (p *fediAPI) CreateLike(ctx context.Context, fMsg messages.FromFediAPI) error { +func (p *fediAPI) CreateLike(ctx context.Context, fMsg *messages.FromFediAPI) error { fave, ok := fMsg.GTSModel.(*gtsmodel.StatusFave) if !ok { return gtserror.Newf("%T not parseable as *gtsmodel.StatusFave", fMsg.GTSModel) @@ -372,7 +359,7 @@ func (p *fediAPI) CreateLike(ctx context.Context, fMsg messages.FromFediAPI) err return nil } -func (p *fediAPI) CreateAnnounce(ctx context.Context, fMsg messages.FromFediAPI) error { +func (p *fediAPI) CreateAnnounce(ctx context.Context, fMsg *messages.FromFediAPI) error { boost, ok := fMsg.GTSModel.(*gtsmodel.Status) if !ok { return gtserror.Newf("%T not parseable as *gtsmodel.Status", fMsg.GTSModel) @@ -386,7 +373,7 @@ func (p *fediAPI) CreateAnnounce(ctx context.Context, fMsg messages.FromFediAPI) boost, err = p.federate.EnrichAnnounce( ctx, boost, - fMsg.ReceivingAccount.Username, + fMsg.Receiving.Username, ) if err != nil { if gtserror.IsUnretrievable(err) { @@ -400,7 +387,7 @@ func (p *fediAPI) CreateAnnounce(ctx context.Context, fMsg messages.FromFediAPI) } // Update stats for the remote account. - if err := p.utilF.incrementStatusesCount(ctx, fMsg.RequestingAccount, boost); err != nil { + if err := p.utils.incrementStatusesCount(ctx, fMsg.Requesting, boost); err != nil { log.Errorf(ctx, "error updating account stats: %v", err) } @@ -420,7 +407,7 @@ func (p *fediAPI) CreateAnnounce(ctx context.Context, fMsg messages.FromFediAPI) return nil } -func (p *fediAPI) CreateBlock(ctx context.Context, fMsg messages.FromFediAPI) error { +func (p *fediAPI) CreateBlock(ctx context.Context, fMsg *messages.FromFediAPI) error { block, ok := fMsg.GTSModel.(*gtsmodel.Block) if !ok { return gtserror.Newf("%T not parseable as *gtsmodel.Block", fMsg.GTSModel) @@ -499,7 +486,7 @@ func (p *fediAPI) CreateBlock(ctx context.Context, fMsg messages.FromFediAPI) er return nil } -func (p *fediAPI) CreateFlag(ctx context.Context, fMsg messages.FromFediAPI) error { +func (p *fediAPI) CreateFlag(ctx context.Context, fMsg *messages.FromFediAPI) error { incomingReport, ok := fMsg.GTSModel.(*gtsmodel.Report) if !ok { return gtserror.Newf("%T not parseable as *gtsmodel.Report", fMsg.GTSModel) @@ -515,7 +502,7 @@ func (p *fediAPI) CreateFlag(ctx context.Context, fMsg messages.FromFediAPI) err return nil } -func (p *fediAPI) UpdateAccount(ctx context.Context, fMsg messages.FromFediAPI) error { +func (p *fediAPI) UpdateAccount(ctx context.Context, fMsg *messages.FromFediAPI) error { // Parse the old/existing account model. account, ok := fMsg.GTSModel.(*gtsmodel.Account) if !ok { @@ -523,15 +510,15 @@ func (p *fediAPI) UpdateAccount(ctx context.Context, fMsg messages.FromFediAPI) } // Because this was an Update, the new Accountable should be set on the message. - apubAcc, ok := fMsg.APObjectModel.(ap.Accountable) + apubAcc, ok := fMsg.APObject.(ap.Accountable) if !ok { - return gtserror.Newf("cannot cast %T -> ap.Accountable", fMsg.APObjectModel) + return gtserror.Newf("cannot cast %T -> ap.Accountable", fMsg.APObject) } // Fetch up-to-date bio, avatar, header, etc. _, _, err := p.federate.RefreshAccount( ctx, - fMsg.ReceivingAccount.Username, + fMsg.Receiving.Username, account, apubAcc, // Force refresh within 5min window. @@ -544,25 +531,25 @@ func (p *fediAPI) UpdateAccount(ctx context.Context, fMsg messages.FromFediAPI) return nil } -func (p *fediAPI) AcceptFollow(ctx context.Context, fMsg messages.FromFediAPI) error { +func (p *fediAPI) AcceptFollow(ctx context.Context, fMsg *messages.FromFediAPI) error { // Update stats for the remote account. - if err := p.utilF.decrementFollowRequestsCount(ctx, fMsg.RequestingAccount); err != nil { + if err := p.utils.decrementFollowRequestsCount(ctx, fMsg.Requesting); err != nil { log.Errorf(ctx, "error updating account stats: %v", err) } - if err := p.utilF.incrementFollowersCount(ctx, fMsg.RequestingAccount); err != nil { + if err := p.utils.incrementFollowersCount(ctx, fMsg.Requesting); err != nil { log.Errorf(ctx, "error updating account stats: %v", err) } // Update stats for the local account. - if err := p.utilF.incrementFollowingCount(ctx, fMsg.ReceivingAccount); err != nil { + if err := p.utils.incrementFollowingCount(ctx, fMsg.Receiving); err != nil { log.Errorf(ctx, "error updating account stats: %v", err) } return nil } -func (p *fediAPI) UpdateStatus(ctx context.Context, fMsg messages.FromFediAPI) error { +func (p *fediAPI) UpdateStatus(ctx context.Context, fMsg *messages.FromFediAPI) error { // Cast the existing Status model attached to msg. existing, ok := fMsg.GTSModel.(*gtsmodel.Status) if !ok { @@ -570,12 +557,12 @@ func (p *fediAPI) UpdateStatus(ctx context.Context, fMsg messages.FromFediAPI) e } // Cast the updated ActivityPub statusable object . - apStatus, _ := fMsg.APObjectModel.(ap.Statusable) + apStatus, _ := fMsg.APObject.(ap.Statusable) // Fetch up-to-date attach status attachments, etc. status, _, err := p.federate.RefreshStatus( ctx, - fMsg.ReceivingAccount.Username, + fMsg.Receiving.Username, existing, apStatus, // Force refresh within 5min window. @@ -605,7 +592,7 @@ func (p *fediAPI) UpdateStatus(ctx context.Context, fMsg messages.FromFediAPI) e return nil } -func (p *fediAPI) DeleteStatus(ctx context.Context, fMsg messages.FromFediAPI) error { +func (p *fediAPI) DeleteStatus(ctx context.Context, fMsg *messages.FromFediAPI) error { // Delete attachments from this status, since this request // comes from the federating API, and there's no way the // poster can do a delete + redraft for it on our instance. @@ -616,12 +603,34 @@ func (p *fediAPI) DeleteStatus(ctx context.Context, fMsg messages.FromFediAPI) e return gtserror.Newf("%T not parseable as *gtsmodel.Status", fMsg.GTSModel) } - if err := p.utilF.wipeStatus(ctx, status, deleteAttachments); err != nil { + // 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) + } + + // 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) + + // First perform the actual status deletion. + if err := p.utils.wipeStatus(ctx, status, deleteAttachments); err != nil { log.Errorf(ctx, "error wiping status: %v", err) } // Update stats for the remote account. - if err := p.utilF.decrementStatusesCount(ctx, fMsg.RequestingAccount); err != nil { + if err := p.utils.decrementStatusesCount(ctx, fMsg.Requesting); err != nil { log.Errorf(ctx, "error updating account stats: %v", err) } @@ -634,12 +643,28 @@ func (p *fediAPI) DeleteStatus(ctx context.Context, fMsg messages.FromFediAPI) e return nil } -func (p *fediAPI) DeleteAccount(ctx context.Context, fMsg messages.FromFediAPI) error { +func (p *fediAPI) DeleteAccount(ctx context.Context, fMsg *messages.FromFediAPI) error { account, ok := fMsg.GTSModel.(*gtsmodel.Account) if !ok { return gtserror.Newf("%T not parseable as *gtsmodel.Account", fMsg.GTSModel) } + // 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) + + // 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) } diff --git a/internal/processing/workers/fromfediapi_move.go b/internal/processing/workers/fromfediapi_move.go index 0188a5d14..d1e43c0c7 100644 --- a/internal/processing/workers/fromfediapi_move.go +++ b/internal/processing/workers/fromfediapi_move.go @@ -220,10 +220,7 @@ func (p *fediAPI) GetOrCreateMove( // APActivityType: "Move" // GTSModel: stub *gtsmodel.Move. // ReceivingAccount: Account of inbox owner receiving the Move. -func (p *fediAPI) MoveAccount(ctx context.Context, fMsg messages.FromFediAPI) error { - // The account who received the Move message. - receiver := fMsg.ReceivingAccount - +func (p *fediAPI) MoveAccount(ctx context.Context, fMsg *messages.FromFediAPI) error { // *gtsmodel.Move activity. stubMove, ok := fMsg.GTSModel.(*gtsmodel.Move) if !ok { @@ -236,7 +233,7 @@ func (p *fediAPI) MoveAccount(ctx context.Context, fMsg messages.FromFediAPI) er // Move origin and target info. var ( originAcctURIStr = stubMove.OriginURI - originAcct = fMsg.RequestingAccount + originAcct = fMsg.Requesting targetAcctURIStr = stubMove.TargetURI targetAcctURI = stubMove.Target ) @@ -308,7 +305,7 @@ func (p *fediAPI) MoveAccount(ctx context.Context, fMsg messages.FromFediAPI) er // Account to which the Move is taking place. targetAcct, targetAcctable, err := p.federate.GetAccountByURI( ctx, - receiver.Username, + fMsg.Receiving.Username, targetAcctURI, ) if err != nil { @@ -340,7 +337,7 @@ func (p *fediAPI) MoveAccount(ctx context.Context, fMsg messages.FromFediAPI) er // Force refresh Move target account // to ensure we have up-to-date version. targetAcct, _, err = p.federate.RefreshAccount(ctx, - receiver.Username, + fMsg.Receiving.Username, targetAcct, targetAcctable, dereferencing.Freshest, @@ -379,7 +376,7 @@ func (p *fediAPI) MoveAccount(ctx context.Context, fMsg messages.FromFediAPI) er // Transfer originAcct's followers // on this instance to targetAcct. - redirectOK := p.utilF.redirectFollowers( + redirectOK := p.utils.redirectFollowers( ctx, originAcct, targetAcct, diff --git a/internal/processing/workers/fromfediapi_test.go b/internal/processing/workers/fromfediapi_test.go index eb3d73e0c..b8acca94d 100644 --- a/internal/processing/workers/fromfediapi_test.go +++ b/internal/processing/workers/fromfediapi_test.go @@ -54,12 +54,12 @@ func (suite *FromFediAPITestSuite) TestProcessFederationAnnounce() { announceStatus.Account = boostingAccount announceStatus.Visibility = boostedStatus.Visibility - err := suite.processor.Workers().ProcessFromFediAPI(context.Background(), messages.FromFediAPI{ - APObjectType: ap.ActivityAnnounce, - APActivityType: ap.ActivityCreate, - GTSModel: announceStatus, - ReceivingAccount: suite.testAccounts["local_account_1"], - RequestingAccount: boostingAccount, + err := suite.processor.Workers().ProcessFromFediAPI(context.Background(), &messages.FromFediAPI{ + APObjectType: ap.ActivityAnnounce, + APActivityType: ap.ActivityCreate, + GTSModel: announceStatus, + Receiving: suite.testAccounts["local_account_1"], + Requesting: boostingAccount, }) suite.NoError(err) @@ -115,12 +115,12 @@ func (suite *FromFediAPITestSuite) TestProcessReplyMention() { suite.NoError(errWithCode) // Send the replied status off to the fedi worker to be further processed. - err = suite.processor.Workers().ProcessFromFediAPI(context.Background(), messages.FromFediAPI{ - APObjectType: ap.ObjectNote, - APActivityType: ap.ActivityCreate, - APObjectModel: replyingStatusable, - ReceivingAccount: repliedAccount, - RequestingAccount: replyingAccount, + err = suite.processor.Workers().ProcessFromFediAPI(context.Background(), &messages.FromFediAPI{ + APObjectType: ap.ObjectNote, + APActivityType: ap.ActivityCreate, + APObject: replyingStatusable, + Receiving: repliedAccount, + Requesting: replyingAccount, }) suite.NoError(err) @@ -179,12 +179,12 @@ func (suite *FromFediAPITestSuite) TestProcessFave() { err := suite.db.Put(context.Background(), fave) suite.NoError(err) - err = suite.processor.Workers().ProcessFromFediAPI(context.Background(), messages.FromFediAPI{ - APObjectType: ap.ActivityLike, - APActivityType: ap.ActivityCreate, - GTSModel: fave, - ReceivingAccount: favedAccount, - RequestingAccount: favingAccount, + err = suite.processor.Workers().ProcessFromFediAPI(context.Background(), &messages.FromFediAPI{ + APObjectType: ap.ActivityLike, + APActivityType: ap.ActivityCreate, + GTSModel: fave, + Receiving: favedAccount, + Requesting: favingAccount, }) suite.NoError(err) @@ -249,12 +249,12 @@ func (suite *FromFediAPITestSuite) TestProcessFaveWithDifferentReceivingAccount( err := suite.db.Put(context.Background(), fave) suite.NoError(err) - err = suite.processor.Workers().ProcessFromFediAPI(context.Background(), messages.FromFediAPI{ - APObjectType: ap.ActivityLike, - APActivityType: ap.ActivityCreate, - GTSModel: fave, - ReceivingAccount: receivingAccount, - RequestingAccount: favingAccount, + err = suite.processor.Workers().ProcessFromFediAPI(context.Background(), &messages.FromFediAPI{ + APObjectType: ap.ActivityLike, + APActivityType: ap.ActivityCreate, + GTSModel: fave, + Receiving: receivingAccount, + Requesting: favingAccount, }) suite.NoError(err) @@ -321,12 +321,12 @@ func (suite *FromFediAPITestSuite) TestProcessAccountDelete() { suite.NoError(err) // now they are mufos! - err = suite.processor.Workers().ProcessFromFediAPI(ctx, messages.FromFediAPI{ - APObjectType: ap.ObjectProfile, - APActivityType: ap.ActivityDelete, - GTSModel: deletedAccount, - ReceivingAccount: receivingAccount, - RequestingAccount: deletedAccount, + err = suite.processor.Workers().ProcessFromFediAPI(ctx, &messages.FromFediAPI{ + APObjectType: ap.ObjectProfile, + APActivityType: ap.ActivityDelete, + GTSModel: deletedAccount, + Receiving: receivingAccount, + Requesting: deletedAccount, }) suite.NoError(err) @@ -402,12 +402,12 @@ func (suite *FromFediAPITestSuite) TestProcessFollowRequestLocked() { err := suite.db.Put(ctx, satanFollowRequestTurtle) suite.NoError(err) - err = suite.processor.Workers().ProcessFromFediAPI(ctx, messages.FromFediAPI{ - APObjectType: ap.ActivityFollow, - APActivityType: ap.ActivityCreate, - GTSModel: satanFollowRequestTurtle, - ReceivingAccount: targetAccount, - RequestingAccount: originAccount, + err = suite.processor.Workers().ProcessFromFediAPI(ctx, &messages.FromFediAPI{ + APObjectType: ap.ActivityFollow, + APActivityType: ap.ActivityCreate, + GTSModel: satanFollowRequestTurtle, + Receiving: targetAccount, + Requesting: originAccount, }) suite.NoError(err) @@ -456,12 +456,12 @@ func (suite *FromFediAPITestSuite) TestProcessFollowRequestUnlocked() { err := suite.db.Put(ctx, satanFollowRequestTurtle) suite.NoError(err) - err = suite.processor.Workers().ProcessFromFediAPI(ctx, messages.FromFediAPI{ - APObjectType: ap.ActivityFollow, - APActivityType: ap.ActivityCreate, - GTSModel: satanFollowRequestTurtle, - ReceivingAccount: targetAccount, - RequestingAccount: originAccount, + err = suite.processor.Workers().ProcessFromFediAPI(ctx, &messages.FromFediAPI{ + APObjectType: ap.ActivityFollow, + APActivityType: ap.ActivityCreate, + GTSModel: satanFollowRequestTurtle, + Receiving: targetAccount, + Requesting: originAccount, }) suite.NoError(err) @@ -532,13 +532,13 @@ func (suite *FromFediAPITestSuite) TestCreateStatusFromIRI() { receivingAccount := suite.testAccounts["local_account_1"] statusCreator := suite.testAccounts["remote_account_2"] - err := suite.processor.Workers().ProcessFromFediAPI(ctx, messages.FromFediAPI{ - APObjectType: ap.ObjectNote, - APActivityType: ap.ActivityCreate, - GTSModel: nil, // gtsmodel is nil because this is a forwarded status -- we want to dereference it using the iri - ReceivingAccount: receivingAccount, - RequestingAccount: statusCreator, - APIri: testrig.URLMustParse("http://example.org/users/Some_User/statuses/afaba698-5740-4e32-a702-af61aa543bc1"), + err := suite.processor.Workers().ProcessFromFediAPI(ctx, &messages.FromFediAPI{ + APObjectType: ap.ObjectNote, + APActivityType: ap.ActivityCreate, + GTSModel: nil, // gtsmodel is nil because this is a forwarded status -- we want to dereference it using the iri + Receiving: receivingAccount, + Requesting: statusCreator, + APIRI: testrig.URLMustParse("http://example.org/users/Some_User/statuses/afaba698-5740-4e32-a702-af61aa543bc1"), }) suite.NoError(err) @@ -585,7 +585,7 @@ func (suite *FromFediAPITestSuite) TestMoveAccount() { } // Process the Move. - err := suite.processor.Workers().ProcessFromFediAPI(ctx, messages.FromFediAPI{ + err := suite.processor.Workers().ProcessFromFediAPI(ctx, &messages.FromFediAPI{ APObjectType: ap.ObjectProfile, APActivityType: ap.ActivityMove, GTSModel: >smodel.Move{ @@ -595,8 +595,8 @@ func (suite *FromFediAPITestSuite) TestMoveAccount() { Target: testrig.URLMustParse(targetAcct.URI), URI: "https://fossbros-anonymous.io/users/foss_satan/moves/01HRA064871MR8HGVSAFJ333GM", }, - ReceivingAccount: receivingAcct, - RequestingAccount: requestingAcct, + Receiving: receivingAcct, + Requesting: requestingAcct, }) suite.NoError(err) diff --git a/internal/processing/workers/util.go b/internal/processing/workers/util.go index cd936f428..a01982e1a 100644 --- a/internal/processing/workers/util.go +++ b/internal/processing/workers/util.go @@ -32,9 +32,9 @@ import ( "github.com/superseriousbusiness/gotosocial/internal/state" ) -// utilF wraps util functions used by both +// util provides util functions used by both // the fromClientAPI and fromFediAPI functions. -type utilF struct { +type utils struct { state *state.State media *media.Processor account *account.Processor @@ -45,7 +45,7 @@ type utilF struct { // used to totally delete a status + all // its attachments, notifications, boosts, // and timeline entries. -func (u *utilF) wipeStatus( +func (u *utils) wipeStatus( ctx context.Context, statusToDelete *gtsmodel.Status, deleteAttachments bool, @@ -152,7 +152,7 @@ func (u *utilF) wipeStatus( // already, and the Move must be valid. // // Return bool will be true if all goes OK. -func (u *utilF) redirectFollowers( +func (u *utils) redirectFollowers( ctx context.Context, originAcct *gtsmodel.Account, targetAcct *gtsmodel.Account, @@ -239,7 +239,7 @@ func (u *utilF) redirectFollowers( return true } -func (u *utilF) incrementStatusesCount( +func (u *utils) incrementStatusesCount( ctx context.Context, account *gtsmodel.Account, status *gtsmodel.Status, @@ -271,7 +271,7 @@ func (u *utilF) incrementStatusesCount( return nil } -func (u *utilF) decrementStatusesCount( +func (u *utils) decrementStatusesCount( ctx context.Context, account *gtsmodel.Account, ) error { @@ -305,7 +305,7 @@ func (u *utilF) decrementStatusesCount( return nil } -func (u *utilF) incrementFollowersCount( +func (u *utils) incrementFollowersCount( ctx context.Context, account *gtsmodel.Account, ) error { @@ -334,7 +334,7 @@ func (u *utilF) incrementFollowersCount( return nil } -func (u *utilF) decrementFollowersCount( +func (u *utils) decrementFollowersCount( ctx context.Context, account *gtsmodel.Account, ) error { @@ -368,7 +368,7 @@ func (u *utilF) decrementFollowersCount( return nil } -func (u *utilF) incrementFollowingCount( +func (u *utils) incrementFollowingCount( ctx context.Context, account *gtsmodel.Account, ) error { @@ -397,7 +397,7 @@ func (u *utilF) incrementFollowingCount( return nil } -func (u *utilF) decrementFollowingCount( +func (u *utils) decrementFollowingCount( ctx context.Context, account *gtsmodel.Account, ) error { @@ -431,7 +431,7 @@ func (u *utilF) decrementFollowingCount( return nil } -func (u *utilF) incrementFollowRequestsCount( +func (u *utils) incrementFollowRequestsCount( ctx context.Context, account *gtsmodel.Account, ) error { @@ -460,7 +460,7 @@ func (u *utilF) incrementFollowRequestsCount( return nil } -func (u *utilF) decrementFollowRequestsCount( +func (u *utils) decrementFollowRequestsCount( ctx context.Context, account *gtsmodel.Account, ) error { diff --git a/internal/processing/workers/workers.go b/internal/processing/workers/workers.go index 8488e501c..1159b61a5 100644 --- a/internal/processing/workers/workers.go +++ b/internal/processing/workers/workers.go @@ -30,9 +30,9 @@ import ( ) type Processor struct { + clientAPI clientAPI + fediAPI fediAPI workers *workers.Workers - clientAPI *clientAPI - fediAPI *fediAPI } func New( @@ -45,6 +45,14 @@ func New( media *media.Processor, stream *stream.Processor, ) Processor { + // Init federate logic + // wrapper struct. + federate := &federate{ + Federator: federator, + state: state, + converter: converter, + } + // Init surface logic // wrapper struct. surface := &surface{ @@ -55,16 +63,8 @@ func New( emailSender: emailSender, } - // Init federate logic - // wrapper struct. - federate := &federate{ - Federator: federator, - state: state, - converter: converter, - } - // Init shared util funcs. - utilF := &utilF{ + utils := &utils{ state: state, media: media, account: account, @@ -73,20 +73,20 @@ func New( return Processor{ workers: &state.Workers, - clientAPI: &clientAPI{ + clientAPI: clientAPI{ state: state, converter: converter, surface: surface, federate: federate, account: account, - utilF: utilF, + utils: utils, }, - fediAPI: &fediAPI{ + fediAPI: fediAPI{ state: state, surface: surface, federate: federate, account: account, - utilF: utilF, + utils: utils, }, } } diff --git a/internal/processing/workers/workers_test.go b/internal/processing/workers/workers_test.go index 3ab19d38d..af1428e64 100644 --- a/internal/processing/workers/workers_test.go +++ b/internal/processing/workers/workers_test.go @@ -127,9 +127,6 @@ func (suite *WorkersTestSuite) SetupTest() { suite.processor = processing.NewProcessor(cleaner.New(&suite.state), suite.typeconverter, suite.federator, suite.oauthServer, suite.mediaManager, &suite.state, suite.emailSender) testrig.StartWorkers(&suite.state, suite.processor.Workers()) - suite.state.Workers.EnqueueClientAPI = suite.processor.Workers().EnqueueClientAPI - suite.state.Workers.EnqueueFediAPI = suite.processor.Workers().EnqueueFediAPI - testrig.StandardDBSetup(suite.db, suite.testAccounts) testrig.StandardStorageSetup(suite.storage, "../../../testrig/media") } diff --git a/internal/queue/simple.go b/internal/queue/simple.go new file mode 100644 index 000000000..9307183f2 --- /dev/null +++ b/internal/queue/simple.go @@ -0,0 +1,141 @@ +// GoToSocial +// Copyright (C) GoToSocial Authors admin@gotosocial.org +// SPDX-License-Identifier: AGPL-3.0-or-later +// +// This program is free software: you can redistribute it and/or modify +// it under the terms of the GNU Affero General Public License as published by +// the Free Software Foundation, either version 3 of the License, or +// (at your option) any later version. +// +// This program is distributed in the hope that it will be useful, +// but WITHOUT ANY WARRANTY; without even the implied warranty of +// MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the +// GNU Affero General Public License for more details. +// +// You should have received a copy of the GNU Affero General Public License +// along with this program. If not, see <http://www.gnu.org/licenses/>. + +package queue + +import ( + "context" + "sync" + + "codeberg.org/gruf/go-list" +) + +// SimpleQueue provides a simple concurrency safe +// queue using generics and a memory pool of list +// elements to reduce overall memory usage. +type SimpleQueue[T any] struct { + l list.List[T] + p []*list.Elem[T] + w chan struct{} + m sync.Mutex +} + +// Push will push given value to the queue. +func (q *SimpleQueue[T]) Push(value T) { + q.m.Lock() + elem := q.alloc() + elem.Value = value + q.l.PushElemFront(elem) + if q.w != nil { + close(q.w) + q.w = nil + } + q.m.Unlock() +} + +// Pop will attempt to pop value from the queue. +func (q *SimpleQueue[T]) Pop() (value T, ok bool) { + q.m.Lock() + if ok = (q.l.Tail != nil); ok { + tail := q.l.Tail + value = tail.Value + q.l.Remove(tail) + q.free(tail) + } + q.m.Unlock() + return +} + +// PopCtx will attempt to pop value from queue, else blocking on context. +func (q *SimpleQueue[T]) PopCtx(ctx context.Context) (value T, ok bool) { + + // Acquire lock. + q.m.Lock() + + var elem *list.Elem[T] + + for { + // Get next elem. + elem = q.l.Tail + if ok = (elem != nil); ok { + break + } + + if q.w == nil { + // Create new wait channel. + q.w = make(chan struct{}) + } + + // Get current + // ch pointer. + ch := q.w + + // Done with lock. + q.m.Unlock() + + select { + // Context canceled. + case <-ctx.Done(): + return + + // Pushed! + case <-ch: + } + + // Relock queue. + q.m.Lock() + } + + // Extract value. + value = elem.Value + + // Remove element. + q.l.Remove(elem) + q.free(elem) + + // Done with lock. + q.m.Unlock() + + return +} + +// Len returns the current length of the queue. +func (q *SimpleQueue[T]) Len() int { + q.m.Lock() + l := q.l.Len() + q.m.Unlock() + return l +} + +// alloc will allocate new list element (relying on memory pool). +func (q *SimpleQueue[T]) alloc() *list.Elem[T] { + if len(q.p) > 0 { + elem := q.p[len(q.p)-1] + q.p = q.p[:len(q.p)-1] + return elem + } + return new(list.Elem[T]) +} + +// free will free list element and release to pool. +func (q *SimpleQueue[T]) free(elem *list.Elem[T]) { + var zero T + elem.Next = nil + elem.Prev = nil + elem.Value = zero + q.p = append(q.p, elem) +} diff --git a/internal/queue/wrappers.go b/internal/queue/struct.go index e07984f84..27139daec 100644 --- a/internal/queue/wrappers.go +++ b/internal/queue/struct.go @@ -18,7 +18,7 @@ package queue import ( - "sync/atomic" + "context" "codeberg.org/gruf/go-structr" ) @@ -26,15 +26,14 @@ import ( // StructQueue wraps a structr.Queue{} to // provide simple index caching by name. type StructQueue[StructType any] struct { - queue structr.Queue[StructType] + queue structr.QueueCtx[StructType] index map[string]*structr.Index - wait atomic.Pointer[chan struct{}] } // Init initializes queue with structr.QueueConfig{}. func (q *StructQueue[T]) Init(config structr.QueueConfig[T]) { q.index = make(map[string]*structr.Index, len(config.Indices)) - q.queue = structr.Queue[T]{} + // q.queue = structr.QueueCtx[T]{} q.queue.Init(config) for _, cfg := range config.Indices { q.index[cfg.Fields] = q.queue.Index(cfg.Fields) @@ -43,13 +42,22 @@ func (q *StructQueue[T]) Init(config structr.QueueConfig[T]) { // Pop: see structr.Queue{}.PopFront(). func (q *StructQueue[T]) Pop() (value T, ok bool) { - return q.queue.PopFront() + values := q.queue.PopFrontN(1) + if ok = (len(values) > 0); !ok { + return + } + value = values[0] + return +} + +// PopCtx: see structr.QueueCtx{}.PopFront(). +func (q *StructQueue[T]) PopCtx(ctx context.Context) (value T, ok bool) { + return q.queue.PopFront(ctx) } -// Push wraps structr.Queue{}.PushBack() to awaken those blocking on <-.Wait(). +// Push: see structr.Queue.PushBack(). func (q *StructQueue[T]) Push(values ...T) { q.queue.PushBack(values...) - q.broadcast() } // Delete pops (and drops!) all queued entries under index with key. @@ -66,31 +74,5 @@ func (q *StructQueue[T]) Len() int { // Wait returns current wait channel, which may be // blocked on to awaken when new value pushed to queue. func (q *StructQueue[T]) Wait() <-chan struct{} { - var ch chan struct{} - - for { - // Get channel ptr. - ptr := q.wait.Load() - if ptr != nil { - return *ptr - } - - if ch == nil { - // Allocate new channel. - ch = make(chan struct{}) - } - - // Try set the new wait channel ptr. - if q.wait.CompareAndSwap(ptr, &ch) { - return ch - } - } -} - -// broadcast safely closes wait channel if -// currently set, releasing waiting goroutines. -func (q *StructQueue[T]) broadcast() { - if ptr := q.wait.Swap(nil); ptr != nil { - close(*ptr) - } + return q.queue.Wait() } diff --git a/internal/transport/delivery/delivery.go b/internal/transport/delivery/delivery.go index 27281399f..286f2abd2 100644 --- a/internal/transport/delivery/delivery.go +++ b/internal/transport/delivery/delivery.go @@ -28,6 +28,7 @@ import ( "github.com/superseriousbusiness/gotosocial/internal/httpclient" "github.com/superseriousbusiness/gotosocial/internal/log" "github.com/superseriousbusiness/gotosocial/internal/queue" + "github.com/superseriousbusiness/gotosocial/internal/util" ) // Delivery wraps an httpclient.Request{} @@ -99,29 +100,51 @@ func (p *WorkerPool) Init(client *httpclient.Client) { } // Start will attempt to start 'n' Worker{}s. -func (p *WorkerPool) Start(n int) (ok bool) { - if ok = (len(p.workers) == 0); ok { - p.workers = make([]*Worker, n) - for i := range p.workers { - p.workers[i] = new(Worker) - p.workers[i].Client = p.Client - p.workers[i].Queue = &p.Queue - ok = p.workers[i].Start() && ok - } +func (p *WorkerPool) Start(n int) { + // Check whether workers are + // set (is already running). + ok := (len(p.workers) > 0) + if ok { + return + } + + // Allocate new workers slice. + p.workers = make([]*Worker, n) + for i := range p.workers { + + // Allocate new Worker{}. + p.workers[i] = new(Worker) + p.workers[i].Client = p.Client + p.workers[i].Queue = &p.Queue + + // Attempt to start worker. + // Return bool not useful + // here, as true = started, + // false = already running. + _ = p.workers[i].Start() } - return } // Stop will attempt to stop contained Worker{}s. -func (p *WorkerPool) Stop() (ok bool) { - if ok = (len(p.workers) > 0); ok { - for i := range p.workers { - ok = p.workers[i].Stop() && ok - p.workers[i] = nil - } - p.workers = p.workers[:0] +func (p *WorkerPool) Stop() { + // Check whether workers are + // set (is currently running). + ok := (len(p.workers) == 0) + if ok { + return } - return + + // Stop all running workers. + for i := range p.workers { + + // return bool not useful + // here, as true = stopped, + // false = never running. + _ = p.workers[i].Stop() + } + + // Unset workers slice. + p.workers = p.workers[:0] } // Worker wraps an httpclient.Client{} to feed @@ -158,23 +181,13 @@ func (w *Worker) run(ctx context.Context) { if w.Client == nil || w.Queue == nil { panic("not yet initialized") } - log.Infof(ctx, "%p: started delivery worker", w) - defer log.Infof(ctx, "%p: stopped delivery worker", w) - for returned := false; !returned; { - func() { - defer func() { - if r := recover(); r != nil { - log.Errorf(ctx, "recovered panic: %v", r) - } - }() - w.process(ctx) - returned = true - }() - } + log.Infof(ctx, "%p: starting worker", w) + defer log.Infof(ctx, "%p: stopped worker", w) + util.Must(func() { w.process(ctx) }) } // process is the main delivery worker processing routine. -func (w *Worker) process(ctx context.Context) { +func (w *Worker) process(ctx context.Context) bool { if w.Client == nil || w.Queue == nil { // we perform this check here just // to ensure the compiler knows these @@ -188,7 +201,7 @@ loop: // Get next delivery. dlv, ok := w.next(ctx) if !ok { - return + return true } // Check whether backoff required. @@ -203,7 +216,7 @@ loop: // Main ctx // cancelled. backoff.Stop() - return + return true case <-w.Queue.Wait(): // A new message was diff --git a/internal/transport/delivery/delivery_test.go b/internal/transport/delivery/delivery_test.go index 852c6f6f3..48831f098 100644 --- a/internal/transport/delivery/delivery_test.go +++ b/internal/transport/delivery/delivery_test.go @@ -32,9 +32,7 @@ func testDeliveryWorkerPool(t *testing.T, sz int, input []*testrequest) { "127.0.0.0/8", }), })) - if !wp.Start(sz) { - t.Fatal("failed starting pool") - } + wp.Start(sz) defer wp.Stop() test(t, &wp.Queue, input) } diff --git a/internal/util/fns.go b/internal/util/fns.go new file mode 100644 index 000000000..53f5ebeed --- /dev/null +++ b/internal/util/fns.go @@ -0,0 +1,68 @@ +// GoToSocial +// Copyright (C) GoToSocial Authors admin@gotosocial.org +// SPDX-License-Identifier: AGPL-3.0-or-later +// +// This program is free software: you can redistribute it and/or modify +// it under the terms of the GNU Affero General Public License as published by +// the Free Software Foundation, either version 3 of the License, or +// (at your option) any later version. +// +// This program is distributed in the hope that it will be useful, +// but WITHOUT ANY WARRANTY; without even the implied warranty of +// MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the +// GNU Affero General Public License for more details. +// +// You should have received a copy of the GNU Affero General Public License +// along with this program. If not, see <http://www.gnu.org/licenses/>. + +package util + +import ( + "fmt" + "os" + "runtime" + + "codeberg.org/gruf/go-errors/v2" +) + +// Must executes 'fn' repeatedly until +// it successfully returns without panic. +func Must(fn func()) { + if fn == nil { + panic("nil func") + } + for !func() (done bool) { + defer func() { + if r := recover(); r != nil { + // Gather calling func frames. + pcs := make([]uintptr, 10) + n := runtime.Callers(3, pcs) + i := runtime.CallersFrames(pcs[:n]) + c := gatherFrames(i, n) + + const msg = "recovered panic: %v\n\n%s\n" + fmt.Fprintf(os.Stderr, msg, r, c.String()) + } + }() + fn() + done = true + return + }() { //nolint + } +} + +// gatherFrames collates runtime frames from a frame iterator. +func gatherFrames(iter *runtime.Frames, n int) errors.Callers { + if iter == nil { + return nil + } + frames := make([]runtime.Frame, 0, n) + for { + f, ok := iter.Next() + if !ok { + break + } + frames = append(frames, f) + } + return frames +} diff --git a/internal/workers/worker_fn.go b/internal/workers/worker_fn.go new file mode 100644 index 000000000..c41bee2b0 --- /dev/null +++ b/internal/workers/worker_fn.go @@ -0,0 +1,142 @@ +// GoToSocial +// Copyright (C) GoToSocial Authors admin@gotosocial.org +// SPDX-License-Identifier: AGPL-3.0-or-later +// +// This program is free software: you can redistribute it and/or modify +// it under the terms of the GNU Affero General Public License as published by +// the Free Software Foundation, either version 3 of the License, or +// (at your option) any later version. +// +// This program is distributed in the hope that it will be useful, +// but WITHOUT ANY WARRANTY; without even the implied warranty of +// MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the +// GNU Affero General Public License for more details. +// +// You should have received a copy of the GNU Affero General Public License +// along with this program. If not, see <http://www.gnu.org/licenses/>. + +package workers + +import ( + "context" + + "codeberg.org/gruf/go-runners" + "github.com/superseriousbusiness/gotosocial/internal/log" + "github.com/superseriousbusiness/gotosocial/internal/queue" + "github.com/superseriousbusiness/gotosocial/internal/util" +) + +// FnWorkerPool wraps multiple FnWorker{}s in +// a singular struct for easy multi start / stop. +type FnWorkerPool struct { + + // Queue is embedded queue.SimpleQueue{} + // passed to each of the pool Worker{}s. + Queue queue.SimpleQueue[func(context.Context)] + + // internal fields. + workers []*FnWorker +} + +// Start will attempt to start 'n' FnWorker{}s. +func (p *FnWorkerPool) Start(n int) { + // Check whether workers are + // set (is already running). + ok := (len(p.workers) > 0) + if ok { + return + } + + // Allocate new workers slice. + p.workers = make([]*FnWorker, n) + for i := range p.workers { + + // Allocate new FnWorker{}. + p.workers[i] = new(FnWorker) + p.workers[i].Queue = &p.Queue + + // Attempt to start worker. + // Return bool not useful + // here, as true = started, + // false = already running. + _ = p.workers[i].Start() + } +} + +// Stop will attempt to stop contained FnWorker{}s. +func (p *FnWorkerPool) Stop() { + // Check whether workers are + // set (is currently running). + ok := (len(p.workers) == 0) + if ok { + return + } + + // Stop all running workers. + for i := range p.workers { + + // return bool not useful + // here, as true = stopped, + // false = never running. + _ = p.workers[i].Stop() + } + + // Unset workers slice. + p.workers = p.workers[:0] +} + +// FnWorker wraps a queue.SimpleQueue{} which +// it feeds from to provide it with function +// tasks to execute. It does so in a single +// goroutine with state management utilities. +type FnWorker struct { + + // Queue is the fn queue that FnWorker + // will feed from for upcoming tasks. + Queue *queue.SimpleQueue[func(context.Context)] + + // internal fields. + service runners.Service +} + +// Start will attempt to start the Worker{}. +func (w *FnWorker) Start() bool { + return w.service.GoRun(w.run) +} + +// Stop will attempt to stop the Worker{}. +func (w *FnWorker) Stop() bool { + return w.service.Stop() +} + +// run wraps process to restart on any panic. +func (w *FnWorker) run(ctx context.Context) { + if w.Queue == nil { + panic("not yet initialized") + } + log.Infof(ctx, "%p: starting worker", w) + defer log.Infof(ctx, "%p: stopped worker", w) + util.Must(func() { w.process(ctx) }) +} + +// process is the main delivery worker processing routine. +func (w *FnWorker) process(ctx context.Context) { + if w.Queue == nil { + // we perform this check here just + // to ensure the compiler knows these + // variables aren't nil in the loop, + // even if already checked by caller. + panic("not yet initialized") + } + + for { + // Block until pop next func. + fn, ok := w.Queue.PopCtx(ctx) + if !ok { + return + } + + // run! + fn(ctx) + } +} diff --git a/internal/workers/worker_msg.go b/internal/workers/worker_msg.go new file mode 100644 index 000000000..1920c964e --- /dev/null +++ b/internal/workers/worker_msg.go @@ -0,0 +1,157 @@ +// GoToSocial +// Copyright (C) GoToSocial Authors admin@gotosocial.org +// SPDX-License-Identifier: AGPL-3.0-or-later +// +// This program is free software: you can redistribute it and/or modify +// it under the terms of the GNU Affero General Public License as published by +// the Free Software Foundation, either version 3 of the License, or +// (at your option) any later version. +// +// This program is distributed in the hope that it will be useful, +// but WITHOUT ANY WARRANTY; without even the implied warranty of +// MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the +// GNU Affero General Public License for more details. +// +// You should have received a copy of the GNU Affero General Public License +// along with this program. If not, see <http://www.gnu.org/licenses/>. + +package workers + +import ( + "context" + + "codeberg.org/gruf/go-runners" + "codeberg.org/gruf/go-structr" + "github.com/superseriousbusiness/gotosocial/internal/log" + "github.com/superseriousbusiness/gotosocial/internal/queue" + "github.com/superseriousbusiness/gotosocial/internal/util" +) + +// MsgWorkerPool wraps multiple MsgWorker{}s in +// a singular struct for easy multi start / stop. +type MsgWorkerPool[Msg any] struct { + + // Process handles queued message types. + Process func(context.Context, Msg) error + + // Queue is embedded queue.StructQueue{} + // passed to each of the pool Worker{}s. + Queue queue.StructQueue[Msg] + + // internal fields. + workers []*MsgWorker[Msg] +} + +// Init will initialize the worker pool queue with given struct indices. +func (p *MsgWorkerPool[T]) Init(indices []structr.IndexConfig) { + p.Queue.Init(structr.QueueConfig[T]{Indices: indices}) +} + +// Start will attempt to start 'n' Worker{}s. +func (p *MsgWorkerPool[T]) Start(n int) { + // Check whether workers are + // set (is already running). + ok := (len(p.workers) > 0) + if ok { + return + } + + // Allocate new msg workers slice. + p.workers = make([]*MsgWorker[T], n) + for i := range p.workers { + + // Allocate new MsgWorker[T]{}. + p.workers[i] = new(MsgWorker[T]) + p.workers[i].Process = p.Process + p.workers[i].Queue = &p.Queue + + // Attempt to start worker. + // Return bool not useful + // here, as true = started, + // false = already running. + _ = p.workers[i].Start() + } +} + +// Stop will attempt to stop contained Worker{}s. +func (p *MsgWorkerPool[T]) Stop() { + // Check whether workers are + // set (is currently running). + ok := (len(p.workers) == 0) + if ok { + return + } + + // Stop all running workers. + for i := range p.workers { + + // return bool not useful + // here, as true = stopped, + // false = never running. + _ = p.workers[i].Stop() + } + + // Unset workers slice. + p.workers = p.workers[:0] +} + +// MsgWorker wraps a processing function to +// feed from a queue.StructQueue{} for messages +// to process. It does so in a single goroutine +// with state management utilities. +type MsgWorker[Msg any] struct { + + // Process handles queued message types. + Process func(context.Context, Msg) error + + // Queue is the Delivery{} message queue + // that delivery worker will feed from. + Queue *queue.StructQueue[Msg] + + // internal fields. + service runners.Service +} + +// Start will attempt to start the Worker{}. +func (w *MsgWorker[T]) Start() bool { + return w.service.GoRun(w.run) +} + +// Stop will attempt to stop the Worker{}. +func (w *MsgWorker[T]) Stop() bool { + return w.service.Stop() +} + +// run wraps process to restart on any panic. +func (w *MsgWorker[T]) run(ctx context.Context) { + if w.Process == nil || w.Queue == nil { + panic("not yet initialized") + } + log.Infof(ctx, "%p: starting worker", w) + defer log.Infof(ctx, "%p: stopped worker", w) + util.Must(func() { w.process(ctx) }) +} + +// process is the main delivery worker processing routine. +func (w *MsgWorker[T]) process(ctx context.Context) { + if w.Process == nil || w.Queue == nil { + // we perform this check here just + // to ensure the compiler knows these + // variables aren't nil in the loop, + // even if already checked by caller. + panic("not yet initialized") + } + + for { + // Block until pop next message. + msg, ok := w.Queue.PopCtx(ctx) + if !ok { + return + } + + // Attempt to process popped message type. + if err := w.Process(ctx, msg); err != nil { + log.Errorf(ctx, "%p: error processing: %v", w, err) + } + } +} diff --git a/internal/workers/workers.go b/internal/workers/workers.go index 17728c255..3f4156841 100644 --- a/internal/workers/workers.go +++ b/internal/workers/workers.go @@ -18,11 +18,8 @@ package workers import ( - "context" - "log" "runtime" - "codeberg.org/gruf/go-runners" "github.com/superseriousbusiness/gotosocial/internal/config" "github.com/superseriousbusiness/gotosocial/internal/messages" "github.com/superseriousbusiness/gotosocial/internal/scheduler" @@ -39,77 +36,49 @@ type Workers struct { // indexed queue of Delivery{} objects. Delivery delivery.WorkerPool - // ClientAPI provides a worker pool that handles both - // incoming client actions, and our own side-effects. - ClientAPI runners.WorkerPool - - // Federator provides a worker pool that handles both - // incoming federated actions, and our own side-effects. - Federator runners.WorkerPool - - // Enqueue functions for clientAPI / federator worker pools, - // these are pointers to Processor{}.Enqueue___() msg functions. - // This prevents dependency cycling as Processor depends on Workers. - EnqueueClientAPI func(context.Context, ...messages.FromClientAPI) - EnqueueFediAPI func(context.Context, ...messages.FromFediAPI) - - // Blocking processing functions for clientAPI / federator. - // These are pointers to Processor{}.Process___() msg functions. - // This prevents dependency cycling as Processor depends on Workers. - // - // Rather than queueing messages for asynchronous processing, these - // functions will process immediately and in a blocking manner, and - // will not use up a worker slot. - // - // As such, you should only call them in special cases where something - // synchronous needs to happen before you can do something else. - ProcessFromClientAPI func(context.Context, messages.FromClientAPI) error - ProcessFromFediAPI func(context.Context, messages.FromFediAPI) error - - // Media manager worker pools. - Media runners.WorkerPool + // Client provides a worker pool that handles + // incoming processing jobs from the client API. + Client MsgWorkerPool[*messages.FromClientAPI] + + // Federator provides a worker pool that handles + // incoming processing jobs from the fedi API. + Federator MsgWorkerPool[*messages.FromFediAPI] + + // Dereference provides a worker pool + // for asynchronous dereferencer jobs. + Dereference FnWorkerPool + + // Media provides a worker pool for + // asynchronous media processing jobs. + Media FnWorkerPool // prevent pass-by-value. _ nocopy } -// Start will start all of the contained -// worker pools (and global scheduler). +// StartScheduler starts the job scheduler. +func (w *Workers) StartScheduler() { + _ = w.Scheduler.Start() // false = already running +} + +// Start will start contained worker pools. func (w *Workers) Start() { - // Get currently set GOMAXPROCS. maxprocs := runtime.GOMAXPROCS(0) - - tryUntil("starting scheduler", 5, w.Scheduler.Start) - - tryUntil("start delivery workerpool", 5, func() bool { - n := config.GetAdvancedSenderMultiplier() - if n < 1 { - // clamp min senders to 1. - return w.Delivery.Start(1) - } - return w.Delivery.Start(n * maxprocs) - }) - - tryUntil("starting client API workerpool", 5, func() bool { - return w.ClientAPI.Start(4*maxprocs, 400*maxprocs) - }) - - tryUntil("starting federator workerpool", 5, func() bool { - return w.Federator.Start(4*maxprocs, 400*maxprocs) - }) - - tryUntil("starting media workerpool", 5, func() bool { - return w.Media.Start(8*maxprocs, 80*maxprocs) - }) + w.Delivery.Start(deliveryWorkers(maxprocs)) + w.Client.Start(4 * maxprocs) + w.Federator.Start(4 * maxprocs) + w.Dereference.Start(4 * maxprocs) + w.Media.Start(8 * maxprocs) } // Stop will stop all of the contained worker pools (and global scheduler). func (w *Workers) Stop() { - tryUntil("stopping scheduler", 5, w.Scheduler.Stop) - tryUntil("stopping delivery workerpool", 5, w.Delivery.Stop) - tryUntil("stopping client API workerpool", 5, w.ClientAPI.Stop) - tryUntil("stopping federator workerpool", 5, w.Federator.Stop) - tryUntil("stopping media workerpool", 5, w.Media.Stop) + _ = w.Scheduler.Stop() // false = not running + w.Delivery.Stop() + w.Client.Stop() + w.Federator.Stop() + w.Dereference.Stop() + w.Media.Stop() } // nocopy when embedded will signal linter to @@ -120,12 +89,11 @@ func (*nocopy) Lock() {} func (*nocopy) Unlock() {} -// tryUntil will attempt to call 'do' for 'count' attempts, before panicking with 'msg'. -func tryUntil(msg string, count int, do func() bool) { - for i := 0; i < count; i++ { - if do() { - return - } +func deliveryWorkers(maxprocs int) int { + n := config.GetAdvancedSenderMultiplier() + if n < 1 { + // clamp to 1 + return 1 } - log.Panicf("failed %s after %d tries", msg, count) + return n * maxprocs } |