mirror of
				https://github.com/superseriousbusiness/gotosocial.git
				synced 2025-10-31 12:22:24 -05: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)
 | |
| }
 |