mirror of
https://codeberg.org/superseriousbusiness/gotosocial.git
synced 2024-12-21 16:34:26 +03:00
c9c0773f2c
* start replacing client + federator + media workers with new worker + queue types * refactor federatingDB.Delete(), drop queued messages when deleting account / status * move all queue purging to the processor workers * undo toolchain updates * code comments, ensure dereferencer worker pool gets started * update gruf libraries in readme * start the job scheduler separately to the worker pools * reshuffle ordering or server.go + remove duplicate worker start / stop * update go-list version * fix vendoring * move queue invalidation to before wipeing / deletion, to ensure queued work not dropped * add logging to worker processing functions in testrig, don't start workers in unexpected places * update go-structr to add (+then rely on) QueueCtx{} type * ensure more worker pools get started properly in tests * fix remaining broken tests relying on worker queue logic * fix account test suite queue popping logic, ensure noop workers do not pull from queue * move back accidentally shuffled account deletion order * ensure error (non nil!!) gets passed in refactored federatingDB{}.Delete() * silently drop deletes from accounts not permitted to * don't warn log on forwarded deletes * make if else clauses easier to parse * use getFederatorMsg() * improved code comment * improved code comment re: requesting account delete checks * remove boolean result from worker start / stop since false = already running or already stopped * remove optional passed-in http.client * remove worker starting from the admin CLI commands (we don't need to handle side-effects) * update prune cli to start scheduler but not all of the workers * fix rebase issues * remove redundant return statements * i'm sorry sir linter
170 lines
4.6 KiB
Go
170 lines
4.6 KiB
Go
// 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 admin
|
|
|
|
import (
|
|
"context"
|
|
"slices"
|
|
"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"
|
|
"github.com/superseriousbusiness/gotosocial/internal/state"
|
|
)
|
|
|
|
func errActionConflict(action *gtsmodel.AdminAction) gtserror.WithCode {
|
|
err := gtserror.NewfAt(
|
|
4, // Include caller's function name.
|
|
"an action (%s) is currently running (duration %s) which conflicts with the attempted action",
|
|
action.Key(), time.Since(action.CreatedAt),
|
|
)
|
|
|
|
const help = "wait until this action is complete and try again"
|
|
return gtserror.NewErrorConflict(err, err.Error(), help)
|
|
}
|
|
|
|
type Actions struct {
|
|
r map[string]*gtsmodel.AdminAction
|
|
state *state.State
|
|
|
|
// Not embedded struct,
|
|
// to shield from access
|
|
// by outside packages.
|
|
m sync.Mutex
|
|
}
|
|
|
|
// Run runs the given admin action by executing the supplied function.
|
|
//
|
|
// Run handles locking, action insertion and updating, so you don't have to!
|
|
//
|
|
// If an action is already running which overlaps/conflicts with the
|
|
// given action, an ErrorWithCode 409 will be returned.
|
|
//
|
|
// If execution of the provided function returns errors, the errors
|
|
// will be updated on the provided admin action in the database.
|
|
func (a *Actions) Run(
|
|
ctx context.Context,
|
|
action *gtsmodel.AdminAction,
|
|
f func(context.Context) gtserror.MultiError,
|
|
) gtserror.WithCode {
|
|
actionKey := action.Key()
|
|
|
|
// LOCK THE MAP HERE, since we're
|
|
// going to do some operations on it.
|
|
a.m.Lock()
|
|
|
|
// Bail if an action with
|
|
// this key is already running.
|
|
running, ok := a.r[actionKey]
|
|
if ok {
|
|
a.m.Unlock()
|
|
return errActionConflict(running)
|
|
}
|
|
|
|
// Action with this key not
|
|
// yet running, create it.
|
|
if err := a.state.DB.PutAdminAction(ctx, action); err != nil {
|
|
err = gtserror.Newf("db error putting admin action %s: %w", actionKey, err)
|
|
|
|
// Don't store in map
|
|
// if there's an error.
|
|
a.m.Unlock()
|
|
return gtserror.NewErrorInternalError(err)
|
|
}
|
|
|
|
// Action was inserted,
|
|
// store in map.
|
|
a.r[actionKey] = action
|
|
|
|
// UNLOCK THE MAP HERE, since
|
|
// we're done modifying it for now.
|
|
a.m.Unlock()
|
|
|
|
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))
|
|
for _, err := range errs {
|
|
action.Errors = append(action.Errors, err.Error())
|
|
}
|
|
}
|
|
|
|
// Action is no longer running:
|
|
// remove from running map.
|
|
a.m.Lock()
|
|
delete(a.r, actionKey)
|
|
a.m.Unlock()
|
|
|
|
// Mark as completed in the db,
|
|
// storing errors for later review.
|
|
action.CompletedAt = time.Now()
|
|
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
|
|
}
|
|
|
|
// GetRunning sounds like a threat, but it actually just
|
|
// returns all of the currently running actions held by
|
|
// the Actions struct, ordered by ID descending.
|
|
func (a *Actions) GetRunning() []*gtsmodel.AdminAction {
|
|
a.m.Lock()
|
|
defer a.m.Unlock()
|
|
|
|
// Assemble all currently running actions.
|
|
running := make([]*gtsmodel.AdminAction, 0, len(a.r))
|
|
for _, action := range a.r {
|
|
running = append(running, action)
|
|
}
|
|
|
|
// Order by ID descending (creation date).
|
|
slices.SortFunc(
|
|
running,
|
|
func(a *gtsmodel.AdminAction, b *gtsmodel.AdminAction) int {
|
|
const k = -1
|
|
switch {
|
|
case a.ID > b.ID:
|
|
return +k
|
|
case a.ID < b.ID:
|
|
return -k
|
|
default:
|
|
return 0
|
|
}
|
|
},
|
|
)
|
|
|
|
return running
|
|
}
|
|
|
|
// TotalRunning is a sequel to the classic
|
|
// 1972 environmental-themed science fiction
|
|
// film Silent Running, starring Bruce Dern.
|
|
func (a *Actions) TotalRunning() int {
|
|
a.m.Lock()
|
|
defer a.m.Unlock()
|
|
|
|
return len(a.r)
|
|
}
|