mirror of
				https://github.com/superseriousbusiness/gotosocial.git
				synced 2025-11-04 05:42:25 -06:00 
			
		
		
		
	* 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)
 | 
						|
}
 |