From 18f26cfbf7f9b36b838c0e8762bfba98c89b9797 Mon Sep 17 00:00:00 2001
From: wxiaoguang <wxiaoguang@gmail.com>
Date: Fri, 26 May 2023 15:31:55 +0800
Subject: [PATCH] Improve queue and logger context (#24924)

Before there was a "graceful function": RunWithShutdownFns, it's mainly
for some modules which doesn't support context.

The old queue system doesn't work well with context, so the old queues
need it.

After the queue refactoring, the new queue works with context well, so,
use Golang context as much as possible, the `RunWithShutdownFns` could
be removed (replaced by RunWithCancel for context cancel mechanism), the
related code could be simplified.

This PR also fixes some legacy queue-init problems, eg:

* typo : archiver: "unable to create codes indexer queue" => "unable to
create repo-archive queue"
* no nil check for failed queues, which causes unfriendly panic

After this PR, many goroutines could have better display name:

![image](https://github.com/go-gitea/gitea/assets/2114189/701b2a9b-8065-4137-aeaa-0bda2b34604a)

![image](https://github.com/go-gitea/gitea/assets/2114189/f1d5f50f-0534-40f0-b0be-f2c9daa5fe92)
---
 modules/graceful/context.go              | 43 --------------
 modules/graceful/manager.go              | 62 ++++-----------------
 modules/graceful/manager_unix.go         |  1 -
 modules/graceful/manager_windows.go      |  1 -
 modules/indexer/code/indexer.go          |  6 +-
 modules/indexer/issues/indexer.go        | 71 +++++++++++-------------
 modules/indexer/stats/queue.go           |  8 +--
 modules/log/event_writer_base.go         | 11 +++-
 modules/log/event_writer_conn_test.go    |  2 +-
 modules/log/init.go                      | 19 ++++++-
 modules/log/logger_impl.go               |  4 +-
 modules/log/logger_test.go               |  8 +--
 modules/log/manager.go                   |  4 +-
 modules/mirror/mirror.go                 |  8 ++-
 modules/notification/ui/ui.go            |  7 ++-
 modules/process/manager.go               | 11 +++-
 modules/queue/manager.go                 | 12 ++--
 modules/queue/manager_test.go            |  6 +-
 modules/queue/workergroup.go             | 24 +++++---
 modules/queue/workerqueue.go             | 33 ++++++-----
 modules/queue/workerqueue_test.go        | 30 ++++------
 routers/web/admin/queue_tester.go        | 15 +++--
 services/actions/init.go                 |  8 ++-
 services/automerge/automerge.go          |  6 +-
 services/mailer/mailer.go                | 12 ++--
 services/pull/check.go                   |  6 +-
 services/pull/check_test.go              | 18 ++----
 services/repository/archiver/archiver.go |  9 ++-
 services/repository/push.go              |  7 +--
 services/task/task.go                    |  9 +--
 services/webhook/deliver.go              |  6 +-
 31 files changed, 204 insertions(+), 263 deletions(-)

diff --git a/modules/graceful/context.go b/modules/graceful/context.go
index 6b5b207ff..4fcbcb04b 100644
--- a/modules/graceful/context.go
+++ b/modules/graceful/context.go
@@ -5,51 +5,8 @@ package graceful
 
 import (
 	"context"
-	"time"
 )
 
-// ChannelContext is a context that wraps a channel and error as a context
-type ChannelContext struct {
-	done <-chan struct{}
-	err  error
-}
-
-// NewChannelContext creates a ChannelContext from a channel and error
-func NewChannelContext(done <-chan struct{}, err error) *ChannelContext {
-	return &ChannelContext{
-		done: done,
-		err:  err,
-	}
-}
-
-// Deadline returns the time when work done on behalf of this context
-// should be canceled. There is no Deadline for a ChannelContext
-func (ctx *ChannelContext) Deadline() (deadline time.Time, ok bool) {
-	return deadline, ok
-}
-
-// Done returns the channel provided at the creation of this context.
-// When closed, work done on behalf of this context should be canceled.
-func (ctx *ChannelContext) Done() <-chan struct{} {
-	return ctx.done
-}
-
-// Err returns nil, if Done is not closed. If Done is closed,
-// Err returns the error provided at the creation of this context
-func (ctx *ChannelContext) Err() error {
-	select {
-	case <-ctx.done:
-		return ctx.err
-	default:
-		return nil
-	}
-}
-
-// Value returns nil for all calls as no values are or can be associated with this context
-func (ctx *ChannelContext) Value(key interface{}) interface{} {
-	return nil
-}
-
 // ShutdownContext returns a context.Context that is Done at shutdown
 // Callers using this context should ensure that they are registered as a running server
 // in order that they are waited for.
diff --git a/modules/graceful/manager.go b/modules/graceful/manager.go
index d32788092..3604c0a3f 100644
--- a/modules/graceful/manager.go
+++ b/modules/graceful/manager.go
@@ -23,6 +23,11 @@ const (
 	stateTerminate
 )
 
+type RunCanceler interface {
+	Run()
+	Cancel()
+}
+
 // There are some places that could inherit sockets:
 //
 // * HTTP or HTTPS main listener
@@ -55,46 +60,19 @@ func InitManager(ctx context.Context) {
 	})
 }
 
-// WithCallback is a runnable to call when the caller has finished
-type WithCallback func(callback func())
-
-// RunnableWithShutdownFns is a runnable with functions to run at shutdown and terminate
-// After the callback to atShutdown is called and is complete, the main function must return.
-// Similarly the callback function provided to atTerminate must return once termination is complete.
-// Please note that use of the atShutdown and atTerminate callbacks will create go-routines that will wait till their respective signals
-// - users must therefore be careful to only call these as necessary.
-type RunnableWithShutdownFns func(atShutdown, atTerminate func(func()))
-
-// RunWithShutdownFns takes a function that has both atShutdown and atTerminate callbacks
-// After the callback to atShutdown is called and is complete, the main function must return.
-// Similarly the callback function provided to atTerminate must return once termination is complete.
-// Please note that use of the atShutdown and atTerminate callbacks will create go-routines that will wait till their respective signals
-// - users must therefore be careful to only call these as necessary.
-func (g *Manager) RunWithShutdownFns(run RunnableWithShutdownFns) {
+// RunWithCancel helps to run a function with a custom context, the Cancel function will be called at shutdown
+// The Cancel function should stop the Run function in predictable time.
+func (g *Manager) RunWithCancel(rc RunCanceler) {
+	g.RunAtShutdown(context.Background(), rc.Cancel)
 	g.runningServerWaitGroup.Add(1)
 	defer g.runningServerWaitGroup.Done()
 	defer func() {
 		if err := recover(); err != nil {
-			log.Critical("PANIC during RunWithShutdownFns: %v\nStacktrace: %s", err, log.Stack(2))
+			log.Critical("PANIC during RunWithCancel: %v\nStacktrace: %s", err, log.Stack(2))
 			g.doShutdown()
 		}
 	}()
-	run(func(atShutdown func()) {
-		g.lock.Lock()
-		defer g.lock.Unlock()
-		g.toRunAtShutdown = append(g.toRunAtShutdown,
-			func() {
-				defer func() {
-					if err := recover(); err != nil {
-						log.Critical("PANIC during RunWithShutdownFns: %v\nStacktrace: %s", err, log.Stack(2))
-						g.doShutdown()
-					}
-				}()
-				atShutdown()
-			})
-	}, func(atTerminate func()) {
-		g.RunAtTerminate(atTerminate)
-	})
+	rc.Run()
 }
 
 // RunWithShutdownContext takes a function that has a context to watch for shutdown.
@@ -151,21 +129,6 @@ func (g *Manager) RunAtShutdown(ctx context.Context, shutdown func()) {
 		})
 }
 
-// RunAtHammer creates a go-routine to run the provided function at shutdown
-func (g *Manager) RunAtHammer(hammer func()) {
-	g.lock.Lock()
-	defer g.lock.Unlock()
-	g.toRunAtHammer = append(g.toRunAtHammer,
-		func() {
-			defer func() {
-				if err := recover(); err != nil {
-					log.Critical("PANIC during RunAtHammer: %v\nStacktrace: %s", err, log.Stack(2))
-				}
-			}()
-			hammer()
-		})
-}
-
 func (g *Manager) doShutdown() {
 	if !g.setStateTransition(stateRunning, stateShuttingDown) {
 		g.DoImmediateHammer()
@@ -206,9 +169,6 @@ func (g *Manager) doHammerTime(d time.Duration) {
 		g.hammerCtxCancel()
 		atHammerCtx := pprof.WithLabels(g.terminateCtx, pprof.Labels("graceful-lifecycle", "post-hammer"))
 		pprof.SetGoroutineLabels(atHammerCtx)
-		for _, fn := range g.toRunAtHammer {
-			go fn()
-		}
 	}
 	g.lock.Unlock()
 }
diff --git a/modules/graceful/manager_unix.go b/modules/graceful/manager_unix.go
index f949abfd2..b1fd6da76 100644
--- a/modules/graceful/manager_unix.go
+++ b/modules/graceful/manager_unix.go
@@ -41,7 +41,6 @@ type Manager struct {
 	terminateWaitGroup     sync.WaitGroup
 
 	toRunAtShutdown  []func()
-	toRunAtHammer    []func()
 	toRunAtTerminate []func()
 }
 
diff --git a/modules/graceful/manager_windows.go b/modules/graceful/manager_windows.go
index f3606084a..f676f86d0 100644
--- a/modules/graceful/manager_windows.go
+++ b/modules/graceful/manager_windows.go
@@ -50,7 +50,6 @@ type Manager struct {
 	shutdownRequested      chan struct{}
 
 	toRunAtShutdown  []func()
-	toRunAtHammer    []func()
 	toRunAtTerminate []func()
 }
 
diff --git a/modules/indexer/code/indexer.go b/modules/indexer/code/indexer.go
index e9b8e7650..f38fd6000 100644
--- a/modules/indexer/code/indexer.go
+++ b/modules/indexer/code/indexer.go
@@ -166,7 +166,7 @@ func Init() {
 		handler := func(items ...*IndexerData) (unhandled []*IndexerData) {
 			idx, err := indexer.get()
 			if idx == nil || err != nil {
-				log.Error("Codes indexer handler: unable to get indexer!")
+				log.Warn("Codes indexer handler: indexer is not ready, retry later.")
 				return items
 			}
 
@@ -201,7 +201,7 @@ func Init() {
 			return unhandled
 		}
 
-		indexerQueue = queue.CreateUniqueQueue("code_indexer", handler)
+		indexerQueue = queue.CreateUniqueQueue(ctx, "code_indexer", handler)
 		if indexerQueue == nil {
 			log.Fatal("Unable to create codes indexer queue")
 		}
@@ -259,7 +259,7 @@ func Init() {
 		indexer.set(rIndexer)
 
 		// Start processing the queue
-		go graceful.GetManager().RunWithShutdownFns(indexerQueue.Run)
+		go graceful.GetManager().RunWithCancel(indexerQueue)
 
 		if populate {
 			go graceful.GetManager().RunWithShutdownContext(populateRepoIndexer)
diff --git a/modules/indexer/issues/indexer.go b/modules/indexer/issues/indexer.go
index 76ff80ffc..f36ea1093 100644
--- a/modules/indexer/issues/indexer.go
+++ b/modules/indexer/issues/indexer.go
@@ -102,7 +102,7 @@ var (
 func InitIssueIndexer(syncReindex bool) {
 	ctx, _, finished := process.GetManager().AddTypedContext(context.Background(), "Service: IssueIndexer", process.SystemProcessType, false)
 
-	waitChannel := make(chan time.Duration, 1)
+	indexerInitWaitChannel := make(chan time.Duration, 1)
 
 	// Create the Queue
 	switch setting.Indexer.IssueType {
@@ -110,7 +110,7 @@ func InitIssueIndexer(syncReindex bool) {
 		handler := func(items ...*IndexerData) (unhandled []*IndexerData) {
 			indexer := holder.get()
 			if indexer == nil {
-				log.Error("Issue indexer handler: unable to get indexer.")
+				log.Warn("Issue indexer handler: indexer is not ready, retry later.")
 				return items
 			}
 			toIndex := make([]*IndexerData, 0, len(items))
@@ -138,15 +138,17 @@ func InitIssueIndexer(syncReindex bool) {
 			return unhandled
 		}
 
-		issueIndexerQueue = queue.CreateSimpleQueue("issue_indexer", handler)
+		issueIndexerQueue = queue.CreateSimpleQueue(ctx, "issue_indexer", handler)
 
 		if issueIndexerQueue == nil {
 			log.Fatal("Unable to create issue indexer queue")
 		}
 	default:
-		issueIndexerQueue = queue.CreateSimpleQueue[*IndexerData]("issue_indexer", nil)
+		issueIndexerQueue = queue.CreateSimpleQueue[*IndexerData](ctx, "issue_indexer", nil)
 	}
 
+	graceful.GetManager().RunAtTerminate(finished)
+
 	// Create the Indexer
 	go func() {
 		pprof.SetGoroutineLabels(ctx)
@@ -178,51 +180,41 @@ func InitIssueIndexer(syncReindex bool) {
 				if issueIndexer != nil {
 					issueIndexer.Close()
 				}
-				finished()
 				log.Info("PID: %d Issue Indexer closed", os.Getpid())
 			})
 			log.Debug("Created Bleve Indexer")
 		case "elasticsearch":
-			graceful.GetManager().RunWithShutdownFns(func(_, atTerminate func(func())) {
-				pprof.SetGoroutineLabels(ctx)
-				issueIndexer, err := NewElasticSearchIndexer(setting.Indexer.IssueConnStr, setting.Indexer.IssueIndexerName)
-				if err != nil {
-					log.Fatal("Unable to initialize Elastic Search Issue Indexer at connection: %s Error: %v", setting.Indexer.IssueConnStr, err)
-				}
-				exist, err := issueIndexer.Init()
-				if err != nil {
-					log.Fatal("Unable to issueIndexer.Init with connection %s Error: %v", setting.Indexer.IssueConnStr, err)
-				}
-				populate = !exist
-				holder.set(issueIndexer)
-				atTerminate(finished)
-			})
+			issueIndexer, err := NewElasticSearchIndexer(setting.Indexer.IssueConnStr, setting.Indexer.IssueIndexerName)
+			if err != nil {
+				log.Fatal("Unable to initialize Elastic Search Issue Indexer at connection: %s Error: %v", setting.Indexer.IssueConnStr, err)
+			}
+			exist, err := issueIndexer.Init()
+			if err != nil {
+				log.Fatal("Unable to issueIndexer.Init with connection %s Error: %v", setting.Indexer.IssueConnStr, err)
+			}
+			populate = !exist
+			holder.set(issueIndexer)
 		case "db":
 			issueIndexer := &DBIndexer{}
 			holder.set(issueIndexer)
-			graceful.GetManager().RunAtTerminate(finished)
 		case "meilisearch":
-			graceful.GetManager().RunWithShutdownFns(func(_, atTerminate func(func())) {
-				pprof.SetGoroutineLabels(ctx)
-				issueIndexer, err := NewMeilisearchIndexer(setting.Indexer.IssueConnStr, setting.Indexer.IssueConnAuth, setting.Indexer.IssueIndexerName)
-				if err != nil {
-					log.Fatal("Unable to initialize Meilisearch Issue Indexer at connection: %s Error: %v", setting.Indexer.IssueConnStr, err)
-				}
-				exist, err := issueIndexer.Init()
-				if err != nil {
-					log.Fatal("Unable to issueIndexer.Init with connection %s Error: %v", setting.Indexer.IssueConnStr, err)
-				}
-				populate = !exist
-				holder.set(issueIndexer)
-				atTerminate(finished)
-			})
+			issueIndexer, err := NewMeilisearchIndexer(setting.Indexer.IssueConnStr, setting.Indexer.IssueConnAuth, setting.Indexer.IssueIndexerName)
+			if err != nil {
+				log.Fatal("Unable to initialize Meilisearch Issue Indexer at connection: %s Error: %v", setting.Indexer.IssueConnStr, err)
+			}
+			exist, err := issueIndexer.Init()
+			if err != nil {
+				log.Fatal("Unable to issueIndexer.Init with connection %s Error: %v", setting.Indexer.IssueConnStr, err)
+			}
+			populate = !exist
+			holder.set(issueIndexer)
 		default:
 			holder.cancel()
 			log.Fatal("Unknown issue indexer type: %s", setting.Indexer.IssueType)
 		}
 
 		// Start processing the queue
-		go graceful.GetManager().RunWithShutdownFns(issueIndexerQueue.Run)
+		go graceful.GetManager().RunWithCancel(issueIndexerQueue)
 
 		// Populate the index
 		if populate {
@@ -232,13 +224,14 @@ func InitIssueIndexer(syncReindex bool) {
 				go graceful.GetManager().RunWithShutdownContext(populateIssueIndexer)
 			}
 		}
-		waitChannel <- time.Since(start)
-		close(waitChannel)
+
+		indexerInitWaitChannel <- time.Since(start)
+		close(indexerInitWaitChannel)
 	}()
 
 	if syncReindex {
 		select {
-		case <-waitChannel:
+		case <-indexerInitWaitChannel:
 		case <-graceful.GetManager().IsShutdown():
 		}
 	} else if setting.Indexer.StartupTimeout > 0 {
@@ -249,7 +242,7 @@ func InitIssueIndexer(syncReindex bool) {
 				timeout += setting.GracefulHammerTime
 			}
 			select {
-			case duration := <-waitChannel:
+			case duration := <-indexerInitWaitChannel:
 				log.Info("Issue Indexer Initialization took %v", duration)
 			case <-graceful.GetManager().IsShutdown():
 				log.Warn("Shutdown occurred before issue index initialisation was complete")
diff --git a/modules/indexer/stats/queue.go b/modules/indexer/stats/queue.go
index 46438925e..d002bd57c 100644
--- a/modules/indexer/stats/queue.go
+++ b/modules/indexer/stats/queue.go
@@ -29,13 +29,11 @@ func handler(items ...int64) []int64 {
 }
 
 func initStatsQueue() error {
-	statsQueue = queue.CreateUniqueQueue("repo_stats_update", handler)
+	statsQueue = queue.CreateUniqueQueue(graceful.GetManager().ShutdownContext(), "repo_stats_update", handler)
 	if statsQueue == nil {
-		return fmt.Errorf("Unable to create repo_stats_update Queue")
+		return fmt.Errorf("unable to create repo_stats_update queue")
 	}
-
-	go graceful.GetManager().RunWithShutdownFns(statsQueue.Run)
-
+	go graceful.GetManager().RunWithCancel(statsQueue)
 	return nil
 }
 
diff --git a/modules/log/event_writer_base.go b/modules/log/event_writer_base.go
index f61d9a7b9..1d45d579c 100644
--- a/modules/log/event_writer_base.go
+++ b/modules/log/event_writer_base.go
@@ -8,6 +8,7 @@ import (
 	"fmt"
 	"io"
 	"regexp"
+	"runtime/pprof"
 	"time"
 )
 
@@ -143,9 +144,17 @@ func eventWriterStartGo(ctx context.Context, w EventWriter, shared bool) {
 	}
 	w.Base().shared = shared
 	w.Base().stopped = make(chan struct{})
+
+	ctxDesc := "Logger: EventWriter: " + w.GetWriterName()
+	if shared {
+		ctxDesc = "Logger: EventWriter (shared): " + w.GetWriterName()
+	}
+	writerCtx, writerCancel := newContext(ctx, ctxDesc)
 	go func() {
+		defer writerCancel()
 		defer close(w.Base().stopped)
-		w.Run(ctx)
+		pprof.SetGoroutineLabels(writerCtx)
+		w.Run(writerCtx)
 	}()
 }
 
diff --git a/modules/log/event_writer_conn_test.go b/modules/log/event_writer_conn_test.go
index e08ec025a..69e87aa8c 100644
--- a/modules/log/event_writer_conn_test.go
+++ b/modules/log/event_writer_conn_test.go
@@ -40,7 +40,7 @@ func TestConnLogger(t *testing.T) {
 	level := INFO
 	flags := LstdFlags | LUTC | Lfuncname
 
-	logger := NewLoggerWithWriters(context.Background(), NewEventWriterConn("test-conn", WriterMode{
+	logger := NewLoggerWithWriters(context.Background(), "test", NewEventWriterConn("test-conn", WriterMode{
 		Level:        level,
 		Prefix:       prefix,
 		Flags:        FlagsFromBits(flags),
diff --git a/modules/log/init.go b/modules/log/init.go
index 798ba8641..38a3ad60a 100644
--- a/modules/log/init.go
+++ b/modules/log/init.go
@@ -4,14 +4,19 @@
 package log
 
 import (
+	"context"
 	"runtime"
 	"strings"
+	"sync/atomic"
 
 	"code.gitea.io/gitea/modules/process"
 	"code.gitea.io/gitea/modules/util/rotatingfilewriter"
 )
 
-var projectPackagePrefix string
+var (
+	projectPackagePrefix string
+	processTraceDisabled atomic.Int64
+)
 
 func init() {
 	_, filename, _, _ := runtime.Caller(0)
@@ -24,6 +29,10 @@ func init() {
 	rotatingfilewriter.ErrorPrintf = FallbackErrorf
 
 	process.Trace = func(start bool, pid process.IDType, description string, parentPID process.IDType, typ string) {
+		// the logger manager has its own mutex lock, so it's safe to use "Load" here
+		if processTraceDisabled.Load() != 0 {
+			return
+		}
 		if start && parentPID != "" {
 			Log(1, TRACE, "Start %s: %s (from %s) (%s)", NewColoredValue(pid, FgHiYellow), description, NewColoredValue(parentPID, FgYellow), NewColoredValue(typ, Reset))
 		} else if start {
@@ -33,3 +42,11 @@ func init() {
 		}
 	}
 }
+
+func newContext(parent context.Context, desc string) (ctx context.Context, cancel context.CancelFunc) {
+	// the "process manager" also calls "log.Trace()" to output logs, so if we want to create new contexts by the manager, we need to disable the trace temporarily
+	processTraceDisabled.Add(1)
+	defer processTraceDisabled.Add(-1)
+	ctx, _, cancel = process.GetManager().AddTypedContext(parent, desc, process.SystemProcessType, false)
+	return ctx, cancel
+}
diff --git a/modules/log/logger_impl.go b/modules/log/logger_impl.go
index c7e8fde3c..abd72d326 100644
--- a/modules/log/logger_impl.go
+++ b/modules/log/logger_impl.go
@@ -228,9 +228,9 @@ func (l *LoggerImpl) GetLevel() Level {
 	return Level(l.level.Load())
 }
 
-func NewLoggerWithWriters(ctx context.Context, writer ...EventWriter) *LoggerImpl {
+func NewLoggerWithWriters(ctx context.Context, name string, writer ...EventWriter) *LoggerImpl {
 	l := &LoggerImpl{}
-	l.ctx, l.ctxCancel = context.WithCancel(ctx)
+	l.ctx, l.ctxCancel = newContext(ctx, "Logger: "+name)
 	l.LevelLogger = BaseLoggerToGeneralLogger(l)
 	l.eventWriters = map[string]EventWriter{}
 	l.syncLevelInternal()
diff --git a/modules/log/logger_test.go b/modules/log/logger_test.go
index a91b4d23a..70222f64f 100644
--- a/modules/log/logger_test.go
+++ b/modules/log/logger_test.go
@@ -53,7 +53,7 @@ func newDummyWriter(name string, level Level, delay time.Duration) *dummyWriter
 }
 
 func TestLogger(t *testing.T) {
-	logger := NewLoggerWithWriters(context.Background())
+	logger := NewLoggerWithWriters(context.Background(), "test")
 
 	dump := logger.DumpWriters()
 	assert.EqualValues(t, 0, len(dump))
@@ -88,7 +88,7 @@ func TestLogger(t *testing.T) {
 }
 
 func TestLoggerPause(t *testing.T) {
-	logger := NewLoggerWithWriters(context.Background())
+	logger := NewLoggerWithWriters(context.Background(), "test")
 
 	w1 := newDummyWriter("dummy-1", DEBUG, 0)
 	logger.AddWriters(w1)
@@ -117,7 +117,7 @@ func (t testLogString) LogString() string {
 }
 
 func TestLoggerLogString(t *testing.T) {
-	logger := NewLoggerWithWriters(context.Background())
+	logger := NewLoggerWithWriters(context.Background(), "test")
 
 	w1 := newDummyWriter("dummy-1", DEBUG, 0)
 	w1.Mode.Colorize = true
@@ -130,7 +130,7 @@ func TestLoggerLogString(t *testing.T) {
 }
 
 func TestLoggerExpressionFilter(t *testing.T) {
-	logger := NewLoggerWithWriters(context.Background())
+	logger := NewLoggerWithWriters(context.Background(), "test")
 
 	w1 := newDummyWriter("dummy-1", DEBUG, 0)
 	w1.Mode.Expression = "foo.*"
diff --git a/modules/log/manager.go b/modules/log/manager.go
index bbaef7eb2..b5d6cbf8e 100644
--- a/modules/log/manager.go
+++ b/modules/log/manager.go
@@ -39,7 +39,7 @@ func (m *LoggerManager) GetLogger(name string) *LoggerImpl {
 
 	logger := m.loggers[name]
 	if logger == nil {
-		logger = NewLoggerWithWriters(m.ctx)
+		logger = NewLoggerWithWriters(m.ctx, name)
 		m.loggers[name] = logger
 		if name == DEFAULT {
 			m.defaultLogger.Store(logger)
@@ -137,6 +137,6 @@ func GetManager() *LoggerManager {
 
 func NewManager() *LoggerManager {
 	m := &LoggerManager{writers: map[string]EventWriter{}, loggers: map[string]*LoggerImpl{}}
-	m.ctx, m.ctxCancel = context.WithCancel(context.Background())
+	m.ctx, m.ctxCancel = newContext(context.Background(), "LoggerManager")
 	return m
 }
diff --git a/modules/mirror/mirror.go b/modules/mirror/mirror.go
index 73e591adb..0d9a62473 100644
--- a/modules/mirror/mirror.go
+++ b/modules/mirror/mirror.go
@@ -33,9 +33,11 @@ func StartSyncMirrors(queueHandle func(data ...*SyncRequest) []*SyncRequest) {
 	if !setting.Mirror.Enabled {
 		return
 	}
-	mirrorQueue = queue.CreateUniqueQueue("mirror", queueHandle)
-
-	go graceful.GetManager().RunWithShutdownFns(mirrorQueue.Run)
+	mirrorQueue = queue.CreateUniqueQueue(graceful.GetManager().ShutdownContext(), "mirror", queueHandle)
+	if mirrorQueue == nil {
+		log.Fatal("Unable to create mirror queue")
+	}
+	go graceful.GetManager().RunWithCancel(mirrorQueue)
 }
 
 // AddPullMirrorToQueue adds repoID to mirror queue
diff --git a/modules/notification/ui/ui.go b/modules/notification/ui/ui.go
index a4576b679..6c5f22c12 100644
--- a/modules/notification/ui/ui.go
+++ b/modules/notification/ui/ui.go
@@ -37,7 +37,10 @@ var _ base.Notifier = &notificationService{}
 // NewNotifier create a new notificationService notifier
 func NewNotifier() base.Notifier {
 	ns := &notificationService{}
-	ns.issueQueue = queue.CreateSimpleQueue("notification-service", handler)
+	ns.issueQueue = queue.CreateSimpleQueue(graceful.GetManager().ShutdownContext(), "notification-service", handler)
+	if ns.issueQueue == nil {
+		log.Fatal("Unable to create notification-service queue")
+	}
 	return ns
 }
 
@@ -51,7 +54,7 @@ func handler(items ...issueNotificationOpts) []issueNotificationOpts {
 }
 
 func (ns *notificationService) Run() {
-	go graceful.GetManager().RunWithShutdownFns(ns.issueQueue.Run)
+	go graceful.GetManager().RunWithCancel(ns.issueQueue) // TODO: using "go" here doesn't seem right, just leave it as old code
 }
 
 func (ns *notificationService) NotifyCreateIssueComment(ctx context.Context, doer *user_model.User, repo *repo_model.Repository,
diff --git a/modules/process/manager.go b/modules/process/manager.go
index fdfca3db7..25d503c59 100644
--- a/modules/process/manager.go
+++ b/modules/process/manager.go
@@ -167,6 +167,7 @@ func (pm *Manager) Add(ctx context.Context, description string, cancel context.C
 
 	pm.processMap[pid] = process
 	pm.mutex.Unlock()
+
 	Trace(true, pid, description, parentPID, processType)
 
 	pprofCtx := pprof.WithLabels(ctx, pprof.Labels(DescriptionPProfLabel, description, PPIDPProfLabel, string(parentPID), PIDPProfLabel, string(pid), ProcessTypePProfLabel, processType))
@@ -200,10 +201,16 @@ func (pm *Manager) nextPID() (start time.Time, pid IDType) {
 }
 
 func (pm *Manager) remove(process *process) {
+	deleted := false
+
 	pm.mutex.Lock()
-	defer pm.mutex.Unlock()
-	if p := pm.processMap[process.PID]; p == process {
+	if pm.processMap[process.PID] == process {
 		delete(pm.processMap, process.PID)
+		deleted = true
+	}
+	pm.mutex.Unlock()
+
+	if deleted {
 		Trace(false, process.PID, process.Description, process.ParentPID, process.Type)
 	}
 }
diff --git a/modules/queue/manager.go b/modules/queue/manager.go
index 95b3bad57..8b964c0c2 100644
--- a/modules/queue/manager.go
+++ b/modules/queue/manager.go
@@ -88,22 +88,22 @@ func (m *Manager) FlushAll(ctx context.Context, timeout time.Duration) error {
 }
 
 // CreateSimpleQueue creates a simple queue from global setting config provider by name
-func CreateSimpleQueue[T any](name string, handler HandlerFuncT[T]) *WorkerPoolQueue[T] {
-	return createWorkerPoolQueue(name, setting.CfgProvider, handler, false)
+func CreateSimpleQueue[T any](ctx context.Context, name string, handler HandlerFuncT[T]) *WorkerPoolQueue[T] {
+	return createWorkerPoolQueue(ctx, name, setting.CfgProvider, handler, false)
 }
 
 // CreateUniqueQueue creates a unique queue from global setting config provider by name
-func CreateUniqueQueue[T any](name string, handler HandlerFuncT[T]) *WorkerPoolQueue[T] {
-	return createWorkerPoolQueue(name, setting.CfgProvider, handler, true)
+func CreateUniqueQueue[T any](ctx context.Context, name string, handler HandlerFuncT[T]) *WorkerPoolQueue[T] {
+	return createWorkerPoolQueue(ctx, name, setting.CfgProvider, handler, true)
 }
 
-func createWorkerPoolQueue[T any](name string, cfgProvider setting.ConfigProvider, handler HandlerFuncT[T], unique bool) *WorkerPoolQueue[T] {
+func createWorkerPoolQueue[T any](ctx context.Context, name string, cfgProvider setting.ConfigProvider, handler HandlerFuncT[T], unique bool) *WorkerPoolQueue[T] {
 	queueSetting, err := setting.GetQueueSettings(cfgProvider, name)
 	if err != nil {
 		log.Error("Failed to get queue settings for %q: %v", name, err)
 		return nil
 	}
-	w, err := NewWorkerPoolQueueBySetting(name, queueSetting, handler, unique)
+	w, err := NewWorkerPoolQueueWithContext(ctx, name, queueSetting, handler, unique)
 	if err != nil {
 		log.Error("Failed to create queue %q: %v", name, err)
 		return nil
diff --git a/modules/queue/manager_test.go b/modules/queue/manager_test.go
index 50265e27b..1fd29f813 100644
--- a/modules/queue/manager_test.go
+++ b/modules/queue/manager_test.go
@@ -29,7 +29,7 @@ func TestManager(t *testing.T) {
 		if err != nil {
 			return nil, err
 		}
-		return NewWorkerPoolQueueBySetting(name, qs, func(s ...int) (unhandled []int) { return nil }, false)
+		return newWorkerPoolQueueForTest(name, qs, func(s ...int) (unhandled []int) { return nil }, false)
 	}
 
 	// test invalid CONN_STR
@@ -80,7 +80,7 @@ MAX_WORKERS = 2
 
 	assert.NoError(t, err)
 
-	q1 := createWorkerPoolQueue[string]("no-such", cfgProvider, nil, false)
+	q1 := createWorkerPoolQueue[string](context.Background(), "no-such", cfgProvider, nil, false)
 	assert.Equal(t, "no-such", q1.GetName())
 	assert.Equal(t, "dummy", q1.GetType()) // no handler, so it becomes dummy
 	assert.Equal(t, filepath.Join(setting.AppDataPath, "queues/dir1"), q1.baseConfig.DataFullDir)
@@ -96,7 +96,7 @@ MAX_WORKERS = 2
 	assert.Equal(t, "string", q1.GetItemTypeName())
 	qid1 := GetManager().qidCounter
 
-	q2 := createWorkerPoolQueue("sub", cfgProvider, func(s ...int) (unhandled []int) { return nil }, false)
+	q2 := createWorkerPoolQueue(context.Background(), "sub", cfgProvider, func(s ...int) (unhandled []int) { return nil }, false)
 	assert.Equal(t, "sub", q2.GetName())
 	assert.Equal(t, "level", q2.GetType())
 	assert.Equal(t, filepath.Join(setting.AppDataPath, "queues/dir2"), q2.baseConfig.DataFullDir)
diff --git a/modules/queue/workergroup.go b/modules/queue/workergroup.go
index 7127ea111..147a4f335 100644
--- a/modules/queue/workergroup.go
+++ b/modules/queue/workergroup.go
@@ -5,6 +5,7 @@ package queue
 
 import (
 	"context"
+	"runtime/pprof"
 	"sync"
 	"sync/atomic"
 	"time"
@@ -13,9 +14,10 @@ import (
 )
 
 var (
-	infiniteTimerC        = make(chan time.Time)
-	batchDebounceDuration = 100 * time.Millisecond
-	workerIdleDuration    = 1 * time.Second
+	infiniteTimerC         = make(chan time.Time)
+	batchDebounceDuration  = 100 * time.Millisecond
+	workerIdleDuration     = 1 * time.Second
+	shutdownDefaultTimeout = 2 * time.Second
 
 	unhandledItemRequeueDuration atomic.Int64 // to avoid data race during test
 )
@@ -116,13 +118,15 @@ func (q *WorkerPoolQueue[T]) doWorkerHandle(batch []T) {
 // If the queue is shutting down, it returns true and try to push the items
 // Otherwise it does nothing and returns false
 func (q *WorkerPoolQueue[T]) basePushForShutdown(items ...T) bool {
-	ctxShutdown := q.ctxShutdown.Load()
-	if ctxShutdown == nil {
+	shutdownTimeout := time.Duration(q.shutdownTimeout.Load())
+	if shutdownTimeout == 0 {
 		return false
 	}
+	ctxShutdown, ctxShutdownCancel := context.WithTimeout(context.Background(), shutdownTimeout)
+	defer ctxShutdownCancel()
 	for _, item := range items {
 		// if there is still any error, the queue can do nothing instead of losing the items
-		if err := q.baseQueue.PushItem(*ctxShutdown, q.marshal(item)); err != nil {
+		if err := q.baseQueue.PushItem(ctxShutdown, q.marshal(item)); err != nil {
 			log.Error("Failed to requeue item for queue %q when shutting down: %v", q.GetName(), err)
 		}
 	}
@@ -246,6 +250,8 @@ var skipFlushChan = make(chan flushType) // an empty flush chan, used to skip re
 
 // doRun is the main loop of the queue. All related "doXxx" functions are executed in its context.
 func (q *WorkerPoolQueue[T]) doRun() {
+	pprof.SetGoroutineLabels(q.ctxRun)
+
 	log.Debug("Queue %q starts running", q.GetName())
 	defer log.Debug("Queue %q stops running", q.GetName())
 
@@ -271,8 +277,8 @@ func (q *WorkerPoolQueue[T]) doRun() {
 			}
 		}
 
-		ctxShutdownPtr := q.ctxShutdown.Load()
-		if ctxShutdownPtr != nil {
+		shutdownTimeout := time.Duration(q.shutdownTimeout.Load())
+		if shutdownTimeout != 0 {
 			// if there is a shutdown context, try to push the items back to the base queue
 			q.basePushForShutdown(unhandled...)
 			workerDone := make(chan struct{})
@@ -280,7 +286,7 @@ func (q *WorkerPoolQueue[T]) doRun() {
 			go func() { wg.wg.Wait(); close(workerDone) }()
 			select {
 			case <-workerDone:
-			case <-(*ctxShutdownPtr).Done():
+			case <-time.After(shutdownTimeout):
 				log.Error("Queue %q is shutting down, but workers are still running after timeout", q.GetName())
 			}
 		} else {
diff --git a/modules/queue/workerqueue.go b/modules/queue/workerqueue.go
index 5695c6cc2..e0d5183bd 100644
--- a/modules/queue/workerqueue.go
+++ b/modules/queue/workerqueue.go
@@ -10,9 +10,9 @@ import (
 	"sync/atomic"
 	"time"
 
-	"code.gitea.io/gitea/modules/graceful"
 	"code.gitea.io/gitea/modules/json"
 	"code.gitea.io/gitea/modules/log"
+	"code.gitea.io/gitea/modules/process"
 	"code.gitea.io/gitea/modules/setting"
 )
 
@@ -21,8 +21,9 @@ import (
 type WorkerPoolQueue[T any] struct {
 	ctxRun       context.Context
 	ctxRunCancel context.CancelFunc
-	ctxShutdown  atomic.Pointer[context.Context]
-	shutdownDone chan struct{}
+
+	shutdownDone    chan struct{}
+	shutdownTimeout atomic.Int64 // in case some buggy handlers (workers) would hang forever, "shutdown" should finish in predictable time
 
 	origHandler HandlerFuncT[T]
 	safeHandler HandlerFuncT[T]
@@ -175,22 +176,19 @@ func (q *WorkerPoolQueue[T]) Has(data T) (bool, error) {
 	return q.baseQueue.HasItem(q.ctxRun, q.marshal(data))
 }
 
-func (q *WorkerPoolQueue[T]) Run(atShutdown, atTerminate func(func())) {
-	atShutdown(func() {
-		// in case some queue handlers are slow or have hanging bugs, at most wait for a short time
-		q.ShutdownWait(1 * time.Second)
-	})
+func (q *WorkerPoolQueue[T]) Run() {
 	q.doRun()
 }
 
+func (q *WorkerPoolQueue[T]) Cancel() {
+	q.ctxRunCancel()
+}
+
 // ShutdownWait shuts down the queue, waits for all workers to finish their jobs, and pushes the unhandled items back to the base queue
 // It waits for all workers (handlers) to finish their jobs, in case some buggy handlers would hang forever, a reasonable timeout is needed
 func (q *WorkerPoolQueue[T]) ShutdownWait(timeout time.Duration) {
-	shutdownCtx, shutdownCtxCancel := context.WithTimeout(context.Background(), timeout)
-	defer shutdownCtxCancel()
-	if q.ctxShutdown.CompareAndSwap(nil, &shutdownCtx) {
-		q.ctxRunCancel()
-	}
+	q.shutdownTimeout.Store(int64(timeout))
+	q.ctxRunCancel()
 	<-q.shutdownDone
 }
 
@@ -207,7 +205,11 @@ func getNewQueueFn(t string) (string, func(cfg *BaseConfig, unique bool) (baseQu
 	}
 }
 
-func NewWorkerPoolQueueBySetting[T any](name string, queueSetting setting.QueueSettings, handler HandlerFuncT[T], unique bool) (*WorkerPoolQueue[T], error) {
+func newWorkerPoolQueueForTest[T any](name string, queueSetting setting.QueueSettings, handler HandlerFuncT[T], unique bool) (*WorkerPoolQueue[T], error) {
+	return NewWorkerPoolQueueWithContext(context.Background(), name, queueSetting, handler, unique)
+}
+
+func NewWorkerPoolQueueWithContext[T any](ctx context.Context, name string, queueSetting setting.QueueSettings, handler HandlerFuncT[T], unique bool) (*WorkerPoolQueue[T], error) {
 	if handler == nil {
 		log.Debug("Use dummy queue for %q because handler is nil and caller doesn't want to process the queue items", name)
 		queueSetting.Type = "dummy"
@@ -224,10 +226,11 @@ func NewWorkerPoolQueueBySetting[T any](name string, queueSetting setting.QueueS
 	}
 	log.Trace("Created queue %q of type %q", name, queueType)
 
-	w.ctxRun, w.ctxRunCancel = context.WithCancel(graceful.GetManager().ShutdownContext())
+	w.ctxRun, _, w.ctxRunCancel = process.GetManager().AddTypedContext(ctx, "Queue: "+w.GetName(), process.SystemProcessType, false)
 	w.batchChan = make(chan []T)
 	w.flushChan = make(chan flushType)
 	w.shutdownDone = make(chan struct{})
+	w.shutdownTimeout.Store(int64(shutdownDefaultTimeout))
 	w.workerMaxNum = queueSetting.MaxWorkers
 	w.batchLength = queueSetting.BatchLength
 
diff --git a/modules/queue/workerqueue_test.go b/modules/queue/workerqueue_test.go
index da9451cd7..e60120162 100644
--- a/modules/queue/workerqueue_test.go
+++ b/modules/queue/workerqueue_test.go
@@ -16,17 +16,9 @@ import (
 )
 
 func runWorkerPoolQueue[T any](q *WorkerPoolQueue[T]) func() {
-	var stop func()
-	started := make(chan struct{})
-	stopped := make(chan struct{})
-	go func() {
-		q.Run(func(f func()) { stop = f; close(started) }, nil)
-		close(stopped)
-	}()
-	<-started
+	go q.Run()
 	return func() {
-		stop()
-		<-stopped
+		q.ShutdownWait(1 * time.Second)
 	}
 }
 
@@ -57,7 +49,7 @@ func TestWorkerPoolQueueUnhandled(t *testing.T) {
 			return unhandled
 		}
 
-		q, _ := NewWorkerPoolQueueBySetting("test-workpoolqueue", queueSetting, handler, false)
+		q, _ := newWorkerPoolQueueForTest("test-workpoolqueue", queueSetting, handler, false)
 		stop := runWorkerPoolQueue(q)
 		for i := 0; i < queueSetting.Length; i++ {
 			testRecorder.Record("push:%v", i)
@@ -145,7 +137,7 @@ func testWorkerPoolQueuePersistence(t *testing.T, queueSetting setting.QueueSett
 			return nil
 		}
 
-		q, _ := NewWorkerPoolQueueBySetting("pr_patch_checker_test", queueSetting, testHandler, true)
+		q, _ := newWorkerPoolQueueForTest("pr_patch_checker_test", queueSetting, testHandler, true)
 		stop := runWorkerPoolQueue(q)
 		for i := 0; i < testCount; i++ {
 			_ = q.Push("task-" + strconv.Itoa(i))
@@ -169,7 +161,7 @@ func testWorkerPoolQueuePersistence(t *testing.T, queueSetting setting.QueueSett
 			return nil
 		}
 
-		q, _ := NewWorkerPoolQueueBySetting("pr_patch_checker_test", queueSetting, testHandler, true)
+		q, _ := newWorkerPoolQueueForTest("pr_patch_checker_test", queueSetting, testHandler, true)
 		stop := runWorkerPoolQueue(q)
 		assert.NoError(t, q.FlushWithContext(context.Background(), 0))
 		stop()
@@ -194,7 +186,7 @@ func TestWorkerPoolQueueActiveWorkers(t *testing.T) {
 		return nil
 	}
 
-	q, _ := NewWorkerPoolQueueBySetting("test-workpoolqueue", setting.QueueSettings{Type: "channel", BatchLength: 1, MaxWorkers: 1, Length: 100}, handler, false)
+	q, _ := newWorkerPoolQueueForTest("test-workpoolqueue", setting.QueueSettings{Type: "channel", BatchLength: 1, MaxWorkers: 1, Length: 100}, handler, false)
 	stop := runWorkerPoolQueue(q)
 	for i := 0; i < 5; i++ {
 		assert.NoError(t, q.Push(i))
@@ -210,7 +202,7 @@ func TestWorkerPoolQueueActiveWorkers(t *testing.T) {
 	assert.EqualValues(t, 1, q.GetWorkerNumber()) // there is at least one worker after the queue begins working
 	stop()
 
-	q, _ = NewWorkerPoolQueueBySetting("test-workpoolqueue", setting.QueueSettings{Type: "channel", BatchLength: 1, MaxWorkers: 3, Length: 100}, handler, false)
+	q, _ = newWorkerPoolQueueForTest("test-workpoolqueue", setting.QueueSettings{Type: "channel", BatchLength: 1, MaxWorkers: 3, Length: 100}, handler, false)
 	stop = runWorkerPoolQueue(q)
 	for i := 0; i < 15; i++ {
 		assert.NoError(t, q.Push(i))
@@ -238,23 +230,23 @@ func TestWorkerPoolQueueShutdown(t *testing.T) {
 		if items[0] == 0 {
 			close(handlerCalled)
 		}
-		time.Sleep(100 * time.Millisecond)
+		time.Sleep(400 * time.Millisecond)
 		return items
 	}
 
 	qs := setting.QueueSettings{Type: "level", Datadir: t.TempDir() + "/queue", BatchLength: 3, MaxWorkers: 4, Length: 20}
-	q, _ := NewWorkerPoolQueueBySetting("test-workpoolqueue", qs, handler, false)
+	q, _ := newWorkerPoolQueueForTest("test-workpoolqueue", qs, handler, false)
 	stop := runWorkerPoolQueue(q)
 	for i := 0; i < qs.Length; i++ {
 		assert.NoError(t, q.Push(i))
 	}
 	<-handlerCalled
-	time.Sleep(50 * time.Millisecond) // wait for a while to make sure all workers are active
+	time.Sleep(200 * time.Millisecond) // wait for a while to make sure all workers are active
 	assert.EqualValues(t, 4, q.GetWorkerActiveNumber())
 	stop() // stop triggers shutdown
 	assert.EqualValues(t, 0, q.GetWorkerActiveNumber())
 
 	// no item was ever handled, so we still get all of them again
-	q, _ = NewWorkerPoolQueueBySetting("test-workpoolqueue", qs, handler, false)
+	q, _ = newWorkerPoolQueueForTest("test-workpoolqueue", qs, handler, false)
 	assert.EqualValues(t, 20, q.GetQueueItemNumber())
 }
diff --git a/routers/web/admin/queue_tester.go b/routers/web/admin/queue_tester.go
index 96373c4d5..8f713b3bb 100644
--- a/routers/web/admin/queue_tester.go
+++ b/routers/web/admin/queue_tester.go
@@ -4,12 +4,13 @@
 package admin
 
 import (
-	gocontext "context"
+	"runtime/pprof"
 	"sync"
 	"time"
 
 	"code.gitea.io/gitea/modules/graceful"
 	"code.gitea.io/gitea/modules/log"
+	"code.gitea.io/gitea/modules/process"
 	"code.gitea.io/gitea/modules/queue"
 	"code.gitea.io/gitea/modules/setting"
 )
@@ -21,6 +22,7 @@ var testQueueOnce sync.Once
 // developers could see the queue length / worker number / items number on the admin page and try to remove the items
 func initTestQueueOnce() {
 	testQueueOnce.Do(func() {
+		ctx, _, finished := process.GetManager().AddTypedContext(graceful.GetManager().ShutdownContext(), "TestQueue", process.SystemProcessType, false)
 		qs := setting.QueueSettings{
 			Name:        "test-queue",
 			Type:        "channel",
@@ -28,7 +30,7 @@ func initTestQueueOnce() {
 			BatchLength: 2,
 			MaxWorkers:  3,
 		}
-		testQueue, err := queue.NewWorkerPoolQueueBySetting("test-queue", qs, func(t ...int64) (unhandled []int64) {
+		testQueue, err := queue.NewWorkerPoolQueueWithContext(ctx, "test-queue", qs, func(t ...int64) (unhandled []int64) {
 			for range t {
 				select {
 				case <-graceful.GetManager().ShutdownContext().Done():
@@ -44,8 +46,11 @@ func initTestQueueOnce() {
 
 		queue.GetManager().AddManagedQueue(testQueue)
 		testQueue.SetWorkerMaxNumber(5)
-		go graceful.GetManager().RunWithShutdownFns(testQueue.Run)
-		go graceful.GetManager().RunWithShutdownContext(func(ctx gocontext.Context) {
+		go graceful.GetManager().RunWithCancel(testQueue)
+		go func() {
+			pprof.SetGoroutineLabels(ctx)
+			defer finished()
+
 			cnt := int64(0)
 			adding := true
 			for {
@@ -67,6 +72,6 @@ func initTestQueueOnce() {
 					}
 				}
 			}
-		})
+		}()
 	})
 }
diff --git a/services/actions/init.go b/services/actions/init.go
index 8a9a30084..26573c168 100644
--- a/services/actions/init.go
+++ b/services/actions/init.go
@@ -5,6 +5,7 @@ package actions
 
 import (
 	"code.gitea.io/gitea/modules/graceful"
+	"code.gitea.io/gitea/modules/log"
 	"code.gitea.io/gitea/modules/notification"
 	"code.gitea.io/gitea/modules/queue"
 	"code.gitea.io/gitea/modules/setting"
@@ -15,8 +16,11 @@ func Init() {
 		return
 	}
 
-	jobEmitterQueue = queue.CreateUniqueQueue("actions_ready_job", jobEmitterQueueHandler)
-	go graceful.GetManager().RunWithShutdownFns(jobEmitterQueue.Run)
+	jobEmitterQueue = queue.CreateUniqueQueue(graceful.GetManager().ShutdownContext(), "actions_ready_job", jobEmitterQueueHandler)
+	if jobEmitterQueue == nil {
+		log.Fatal("Unable to create actions_ready_job queue")
+	}
+	go graceful.GetManager().RunWithCancel(jobEmitterQueue)
 
 	notification.RegisterNotifier(NewNotifier())
 }
diff --git a/services/automerge/automerge.go b/services/automerge/automerge.go
index f001a6ccc..bf713c443 100644
--- a/services/automerge/automerge.go
+++ b/services/automerge/automerge.go
@@ -29,11 +29,11 @@ var prAutoMergeQueue *queue.WorkerPoolQueue[string]
 
 // Init runs the task queue to that handles auto merges
 func Init() error {
-	prAutoMergeQueue = queue.CreateUniqueQueue("pr_auto_merge", handler)
+	prAutoMergeQueue = queue.CreateUniqueQueue(graceful.GetManager().ShutdownContext(), "pr_auto_merge", handler)
 	if prAutoMergeQueue == nil {
-		return fmt.Errorf("Unable to create pr_auto_merge Queue")
+		return fmt.Errorf("unable to create pr_auto_merge queue")
 	}
-	go graceful.GetManager().RunWithShutdownFns(prAutoMergeQueue.Run)
+	go graceful.GetManager().RunWithCancel(prAutoMergeQueue)
 	return nil
 }
 
diff --git a/services/mailer/mailer.go b/services/mailer/mailer.go
index 5aeda9ed7..ee4721d43 100644
--- a/services/mailer/mailer.go
+++ b/services/mailer/mailer.go
@@ -401,7 +401,9 @@ func NewContext(ctx context.Context) {
 		Sender = &smtpSender{}
 	}
 
-	mailQueue = queue.CreateSimpleQueue("mail", func(items ...*Message) []*Message {
+	subjectTemplates, bodyTemplates = templates.Mailer(ctx)
+
+	mailQueue = queue.CreateSimpleQueue(graceful.GetManager().ShutdownContext(), "mail", func(items ...*Message) []*Message {
 		for _, msg := range items {
 			gomailMsg := msg.ToMessage()
 			log.Trace("New e-mail sending request %s: %s", gomailMsg.GetHeader("To"), msg.Info)
@@ -413,10 +415,10 @@ func NewContext(ctx context.Context) {
 		}
 		return nil
 	})
-
-	go graceful.GetManager().RunWithShutdownFns(mailQueue.Run)
-
-	subjectTemplates, bodyTemplates = templates.Mailer(ctx)
+	if mailQueue == nil {
+		log.Fatal("Unable to create mail queue")
+	}
+	go graceful.GetManager().RunWithCancel(mailQueue)
 }
 
 // SendAsync send mail asynchronously
diff --git a/services/pull/check.go b/services/pull/check.go
index 8bc2bdff1..b5150ad9a 100644
--- a/services/pull/check.go
+++ b/services/pull/check.go
@@ -384,13 +384,13 @@ func CheckPRsForBaseBranch(baseRepo *repo_model.Repository, baseBranchName strin
 
 // Init runs the task queue to test all the checking status pull requests
 func Init() error {
-	prPatchCheckerQueue = queue.CreateUniqueQueue("pr_patch_checker", handler)
+	prPatchCheckerQueue = queue.CreateUniqueQueue(graceful.GetManager().ShutdownContext(), "pr_patch_checker", handler)
 
 	if prPatchCheckerQueue == nil {
-		return fmt.Errorf("Unable to create pr_patch_checker Queue")
+		return fmt.Errorf("unable to create pr_patch_checker queue")
 	}
 
-	go graceful.GetManager().RunWithShutdownFns(prPatchCheckerQueue.Run)
+	go graceful.GetManager().RunWithCancel(prPatchCheckerQueue)
 	go graceful.GetManager().RunWithShutdownContext(InitializePullRequests)
 	return nil
 }
diff --git a/services/pull/check_test.go b/services/pull/check_test.go
index 52209b4d3..57ccb2003 100644
--- a/services/pull/check_test.go
+++ b/services/pull/check_test.go
@@ -5,6 +5,7 @@
 package pull
 
 import (
+	"context"
 	"strconv"
 	"testing"
 	"time"
@@ -31,7 +32,7 @@ func TestPullRequest_AddToTaskQueue(t *testing.T) {
 
 	cfg, err := setting.GetQueueSettings(setting.CfgProvider, "pr_patch_checker")
 	assert.NoError(t, err)
-	prPatchCheckerQueue, err = queue.NewWorkerPoolQueueBySetting("pr_patch_checker", cfg, testHandler, true)
+	prPatchCheckerQueue, err = queue.NewWorkerPoolQueueWithContext(context.Background(), "pr_patch_checker", cfg, testHandler, true)
 	assert.NoError(t, err)
 
 	pr := unittest.AssertExistsAndLoadBean(t, &issues_model.PullRequest{ID: 2})
@@ -46,12 +47,7 @@ func TestPullRequest_AddToTaskQueue(t *testing.T) {
 	assert.True(t, has)
 	assert.NoError(t, err)
 
-	var queueShutdown, queueTerminate []func()
-	go prPatchCheckerQueue.Run(func(shutdown func()) {
-		queueShutdown = append(queueShutdown, shutdown)
-	}, func(terminate func()) {
-		queueTerminate = append(queueTerminate, terminate)
-	})
+	go prPatchCheckerQueue.Run()
 
 	select {
 	case id := <-idChan:
@@ -67,12 +63,6 @@ func TestPullRequest_AddToTaskQueue(t *testing.T) {
 	pr = unittest.AssertExistsAndLoadBean(t, &issues_model.PullRequest{ID: 2})
 	assert.Equal(t, issues_model.PullRequestStatusChecking, pr.Status)
 
-	for _, callback := range queueShutdown {
-		callback()
-	}
-	for _, callback := range queueTerminate {
-		callback()
-	}
-
+	prPatchCheckerQueue.ShutdownWait(5 * time.Second)
 	prPatchCheckerQueue = nil
 }
diff --git a/services/repository/archiver/archiver.go b/services/repository/archiver/archiver.go
index 1c514a411..2e3defee8 100644
--- a/services/repository/archiver/archiver.go
+++ b/services/repository/archiver/archiver.go
@@ -297,7 +297,7 @@ func ArchiveRepository(request *ArchiveRequest) (*repo_model.RepoArchiver, error
 
 var archiverQueue *queue.WorkerPoolQueue[*ArchiveRequest]
 
-// Init initlize archive
+// Init initializes archiver
 func Init() error {
 	handler := func(items ...*ArchiveRequest) []*ArchiveRequest {
 		for _, archiveReq := range items {
@@ -309,12 +309,11 @@ func Init() error {
 		return nil
 	}
 
-	archiverQueue = queue.CreateUniqueQueue("repo-archive", handler)
+	archiverQueue = queue.CreateUniqueQueue(graceful.GetManager().ShutdownContext(), "repo-archive", handler)
 	if archiverQueue == nil {
-		return errors.New("unable to create codes indexer queue")
+		return errors.New("unable to create repo-archive queue")
 	}
-
-	go graceful.GetManager().RunWithShutdownFns(archiverQueue.Run)
+	go graceful.GetManager().RunWithCancel(archiverQueue)
 
 	return nil
 }
diff --git a/services/repository/push.go b/services/repository/push.go
index f21394891..571eedccb 100644
--- a/services/repository/push.go
+++ b/services/repository/push.go
@@ -42,12 +42,11 @@ func handler(items ...[]*repo_module.PushUpdateOptions) [][]*repo_module.PushUpd
 }
 
 func initPushQueue() error {
-	pushQueue = queue.CreateSimpleQueue("push_update", handler)
+	pushQueue = queue.CreateSimpleQueue(graceful.GetManager().ShutdownContext(), "push_update", handler)
 	if pushQueue == nil {
-		return errors.New("unable to create push_update Queue")
+		return errors.New("unable to create push_update queue")
 	}
-
-	go graceful.GetManager().RunWithShutdownFns(pushQueue.Run)
+	go graceful.GetManager().RunWithCancel(pushQueue)
 	return nil
 }
 
diff --git a/services/task/task.go b/services/task/task.go
index 493586a64..11a47a68b 100644
--- a/services/task/task.go
+++ b/services/task/task.go
@@ -37,14 +37,11 @@ func Run(t *admin_model.Task) error {
 
 // Init will start the service to get all unfinished tasks and run them
 func Init() error {
-	taskQueue = queue.CreateSimpleQueue("task", handler)
-
+	taskQueue = queue.CreateSimpleQueue(graceful.GetManager().ShutdownContext(), "task", handler)
 	if taskQueue == nil {
-		return fmt.Errorf("Unable to create Task Queue")
+		return fmt.Errorf("unable to create task queue")
 	}
-
-	go graceful.GetManager().RunWithShutdownFns(taskQueue.Run)
-
+	go graceful.GetManager().RunWithCancel(taskQueue)
 	return nil
 }
 
diff --git a/services/webhook/deliver.go b/services/webhook/deliver.go
index e817783e5..fd7a3d7fb 100644
--- a/services/webhook/deliver.go
+++ b/services/webhook/deliver.go
@@ -283,11 +283,11 @@ func Init() error {
 		},
 	}
 
-	hookQueue = queue.CreateUniqueQueue("webhook_sender", handler)
+	hookQueue = queue.CreateUniqueQueue(graceful.GetManager().ShutdownContext(), "webhook_sender", handler)
 	if hookQueue == nil {
-		return fmt.Errorf("Unable to create webhook_sender Queue")
+		return fmt.Errorf("unable to create webhook_sender queue")
 	}
-	go graceful.GetManager().RunWithShutdownFns(hookQueue.Run)
+	go graceful.GetManager().RunWithCancel(hookQueue)
 
 	go graceful.GetManager().RunWithShutdownContext(populateWebhookSendingQueue)