Cleanup e2e test setup (#6480)

agent/rpc/dial.go

@@ -0,0 +1,112 @@
+// Copyright 2026 Woodpecker Authors
+//
+// Licensed under the Apache License, Version 2.0 (the "License");
+// you may not use this file except in compliance with the License.
+// You may obtain a copy of the License at
+//
+//      http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing, software
+// distributed under the License is distributed on an "AS IS" BASIS,
+// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+// See the License for the specific language governing permissions and
+// limitations under the License.
+
+package rpc
+
+import (
+	"context"
+	"crypto/tls"
+	"fmt"
+	"time"
+
+	"google.golang.org/grpc"
+	grpc_credentials "google.golang.org/grpc/credentials"
+	"google.golang.org/grpc/credentials/insecure"
+	"google.golang.org/grpc/keepalive"
+)
+
+// DialConfig bundles everything Dial needs. Callers build this from a
+// *cli.Command (production) or with literals (tests).
+type DialConfig struct {
+	ServerAddr       string
+	AgentToken       string
+	AgentID          int64
+	Secure           bool
+	SkipTLSVerify    bool
+	KeepaliveTime    time.Duration
+	KeepaliveTimeout time.Duration
+	AuthRefreshEvery time.Duration
+}
+
+// AgentConn holds the two gRPC connections and the auth interceptor an agent
+// needs. Callers are responsible for closing both connections and canceling
+// the auth context passed to Dial when the agent shuts down.
+type AgentConn struct {
+	AuthConn        *grpc.ClientConn
+	MainConn        *grpc.ClientConn
+	AuthInterceptor *AuthInterceptor
+	AgentID         int64
+}
+
+// Close closes both connections. Safe to call even if one or both are nil.
+func (c *AgentConn) Close() {
+	if c.MainConn != nil {
+		_ = c.MainConn.Close()
+	}
+	if c.AuthConn != nil {
+		_ = c.AuthConn.Close()
+	}
+}
+
+// Dial builds the auth gRPC connection, authenticates, then builds the
+// authenticated main gRPC connection.
+//
+// The authCtx parameter governs the lifetime of the token-refresh goroutine
+// inside the interceptor; callers typically want a context separate from
+// request ctx so the interceptor survives long-running polls.
+func Dial(authCtx context.Context, cfg DialConfig) (*AgentConn, error) {
+	var transport grpc.DialOption
+	if cfg.Secure {
+		transport = grpc.WithTransportCredentials(grpc_credentials.NewTLS(
+			&tls.Config{InsecureSkipVerify: cfg.SkipTLSVerify}, //nolint:gosec // user-opt-in via DialConfig.SkipTLSVerify
+		))
+	} else {
+		transport = grpc.WithTransportCredentials(insecure.NewCredentials())
+	}
+
+	keepaliveOpts := grpc.WithKeepaliveParams(keepalive.ClientParameters{
+		Time:    cfg.KeepaliveTime,
+		Timeout: cfg.KeepaliveTimeout,
+	})
+
+	authConn, err := grpc.NewClient(cfg.ServerAddr, transport, keepaliveOpts)
+	if err != nil {
+		return nil, fmt.Errorf("create auth gRPC connection: %w", err)
+	}
+
+	authClient := NewAuthGrpcClient(authConn, cfg.AgentToken, cfg.AgentID)
+	reportedAgentID := authClient.AgentID()
+	authInterceptor, err := NewAuthInterceptor(authCtx, authClient, cfg.AuthRefreshEvery)
+	if err != nil {
+		_ = authConn.Close()
+		return nil, fmt.Errorf("authenticate with server: %w", err)
+	}
+
+	mainConn, err := grpc.NewClient(
+		cfg.ServerAddr, transport, keepaliveOpts,
+		grpc.WithUnaryInterceptor(authInterceptor.Unary()),
+		grpc.WithStreamInterceptor(authInterceptor.Stream()),
+	)
+	if err != nil {
+		_ = authConn.Close()
+		return nil, fmt.Errorf("create main gRPC connection: %w", err)
+	}
+
+	return &AgentConn{
+		AuthConn:        authConn,
+		MainConn:        mainConn,
+		AuthInterceptor: authInterceptor,
+		AgentID:         reportedAgentID,
+	}, nil
+}

cmd/agent/core/agent.go

@@ -17,7 +17,6 @@ package core
 
 import (
 	"context"
-	"crypto/tls"
 	"errors"
 	"fmt"
 	"maps"
@@ -30,11 +29,7 @@ import (
 	"github.com/rs/zerolog/log"
 	"github.com/urfave/cli/v3"
 	"golang.org/x/sync/errgroup"
-	"google.golang.org/grpc"
 	"google.golang.org/grpc/codes"
-	grpc_credentials "google.golang.org/grpc/credentials"
-	"google.golang.org/grpc/credentials/insecure"
-	"google.golang.org/grpc/keepalive"
 	"google.golang.org/grpc/metadata"
 	"google.golang.org/grpc/status"
 
@@ -106,64 +101,41 @@ func run(ctx context.Context, c *cli.Command, backends []types.Backend) error {
 		)
 	}
 
-	var transport grpc.DialOption
-	if c.Bool("grpc-secure") {
-		log.Trace().Msg("use ssl for grpc")
-		transport = grpc.WithTransportCredentials(grpc_credentials.NewTLS(&tls.Config{InsecureSkipVerify: c.Bool("grpc-skip-insecure")}))
-	} else {
-		transport = grpc.WithTransportCredentials(insecure.NewCredentials())
-	}
-
-	authConn, err := grpc.NewClient(
-		c.String("server"),
-		transport,
-		grpc.WithKeepaliveParams(keepalive.ClientParameters{
-			Time:    c.Duration("grpc-keepalive-time"),
-			Timeout: c.Duration("grpc-keepalive-timeout"),
-		}),
-	)
-	if err != nil {
-		return fmt.Errorf("could not create new gRPC 'channel' for authentication: %w", err)
-	}
-	defer authConn.Close()
-
 	agentConfig := readAgentConfig(agentConfigPath)
 
-	agentToken := c.String("grpc-token")
 	grpcClientCtx, grpcClientCtxCancel := context.WithCancelCause(context.Background())
 	defer grpcClientCtxCancel(nil)
-	authClient := agent_rpc.NewAuthGrpcClient(authConn, agentToken, agentConfig.AgentID)
-	authInterceptor, err := agent_rpc.NewAuthInterceptor(grpcClientCtx, authClient, authInterceptorRefreshInterval) //nolint:contextcheck
-	if err != nil {
-		return fmt.Errorf("agent could not auth: %w", err)
+
+	if c.Bool("grpc-secure") {
+		log.Trace().Msg("use ssl for grpc")
 	}
 
+	agentConn, err := agent_rpc.Dial(grpcClientCtx, agent_rpc.DialConfig{ //nolint:contextcheck
+		ServerAddr:       c.String("server"),
+		AgentToken:       c.String("grpc-token"),
+		AgentID:          agentConfig.AgentID,
+		Secure:           c.Bool("grpc-secure"),
+		SkipTLSVerify:    c.Bool("grpc-skip-insecure"),
+		KeepaliveTime:    c.Duration("grpc-keepalive-time"),
+		KeepaliveTimeout: c.Duration("grpc-keepalive-timeout"),
+		AuthRefreshEvery: authInterceptorRefreshInterval,
+	})
+	if err != nil {
+		return err
+	}
+	defer agentConn.Close()
+
 	// Persist the agent ID received during auth so that crashloops reuse the
 	// same server-side entry instead of creating a new one on every restart.
 	if agentConfigPath != "" {
-		agentConfig.AgentID = authClient.AgentID()
+		agentConfig.AgentID = agentConn.AgentID
 		if err := writeAgentConfig(agentConfig, agentConfigPath); err == nil {
 			log.Debug().Msgf("persisted agent ID %d after auth", agentConfig.AgentID)
 		}
 	}
 
-	conn, err := grpc.NewClient(
-		c.String("server"),
-		transport,
-		grpc.WithKeepaliveParams(keepalive.ClientParameters{
-			Time:    c.Duration("grpc-keepalive-time"),
-			Timeout: c.Duration("grpc-keepalive-timeout"),
-		}),
-		grpc.WithUnaryInterceptor(authInterceptor.Unary()),
-		grpc.WithStreamInterceptor(authInterceptor.Stream()),
-	)
-	if err != nil {
-		return fmt.Errorf("could not create new gRPC 'channel' for normal orchestration: %w", err)
-	}
-	defer conn.Close()
-
 	client := agent_rpc.NewGrpcClient(
-		ctx, conn,
+		ctx, agentConn.MainConn,
 		agent_rpc.SetConnectionRetryTimeout(c.Duration("retry-timeout")),
 	)
 	agentConfigPersisted := atomic.Bool{}

cmd/server/grpc_server.go

@@ -19,12 +19,9 @@ import (
 	"fmt"
 	"net"
 
-	"github.com/rs/zerolog/log"
+	"github.com/prometheus/client_golang/prometheus"
 	"github.com/urfave/cli/v3"
-	"google.golang.org/grpc"
-	"google.golang.org/grpc/keepalive"
 
-	"go.woodpecker-ci.org/woodpecker/v3/rpc/proto"
 	"go.woodpecker-ci.org/woodpecker/v3/server"
 	server_rpc "go.woodpecker-ci.org/woodpecker/v3/server/rpc"
 	"go.woodpecker-ci.org/woodpecker/v3/server/store"
@@ -36,52 +33,14 @@ func runGrpcServer(ctx context.Context, c *cli.Command, _store store.Store) erro
 		return fmt.Errorf("failed to listen on grpc-addr: %w", err)
 	}
 
-	jwtSecret := c.String("grpc-secret")
-	jwtManager := server_rpc.NewJWTManager(jwtSecret)
-
-	authorizer := server_rpc.NewAuthorizer(jwtManager)
-	grpcServer := grpc.NewServer(
-		grpc.StreamInterceptor(authorizer.StreamInterceptor),
-		grpc.UnaryInterceptor(authorizer.UnaryInterceptor),
-		grpc.KeepaliveEnforcementPolicy(keepalive.EnforcementPolicy{
-			MinTime: c.Duration("keepalive-min-time"),
-		}),
-	)
-
-	woodpeckerServer := server_rpc.NewWoodpeckerServer(
-		server.Config.Services.Scheduler,
-		server.Config.Services.Logs,
-		_store,
-	)
-	proto.RegisterWoodpeckerServer(grpcServer, woodpeckerServer)
-
-	woodpeckerAuthServer := server_rpc.NewWoodpeckerAuthServer(
-		jwtManager,
-		server.Config.Server.AgentToken,
-		_store,
-	)
-	proto.RegisterWoodpeckerAuthServer(grpcServer, woodpeckerAuthServer)
-
-	grpcCtx, cancel := context.WithCancelCause(ctx)
-	defer cancel(nil)
-
-	go func() {
-		<-grpcCtx.Done()
-		if grpcServer == nil {
-			return
-		}
-		log.Info().Msg("terminating grpc service gracefully")
-		grpcServer.GracefulStop()
-		log.Info().Msg("grpc service stopped")
-	}()
-
-	if err := grpcServer.Serve(lis); err != nil {
-		// signal that we don't have to stop the server gracefully anymore
-		grpcServer = nil
-
-		// wrap the error so we know where it did come from
-		return fmt.Errorf("grpc server failed: %w", err)
-	}
-
-	return nil
+	return server_rpc.Serve(ctx, server_rpc.ServeConfig{
+		Listener:         lis,
+		Store:            _store,
+		Scheduler:        server.Config.Services.Scheduler,
+		Logger:           server.Config.Services.Logs,
+		JWTSecret:        c.String("grpc-secret"),
+		AgentToken:       server.Config.Server.AgentToken,
+		KeepaliveMinTime: c.Duration("keepalive-min-time"),
+		Registerer:       prometheus.DefaultRegisterer,
+	})
 }

e2e/scenarios/fixtures.go

@@ -137,9 +137,8 @@ func LoadScenarios(t *testing.T) []Scenario {
 			{Name: ".woodpecker.yaml", Data: yamlData},
 		}
 
-		if s.Event == "" {
-			s.Event = model.EventPush
-		}
+		s.check(t, stem+".json")
+
 		scenarios = append(scenarios, s)
 	}
 
@@ -178,9 +177,9 @@ func loadMultiWorkflowScenario(t *testing.T, dirName string) Scenario {
 			})
 		}
 	}
-
 	require.NotEmpty(t, files, "no YAML files in multi-workflow dir %s", dirName)
-	require.NotEmpty(t, s.Name, "scenario.json missing 'name' in %s", dirName)
+
+	s.check(t, dirName+"/scenario.json")
 
 	s.Files = files
 	if s.Event == "" {
@@ -188,3 +187,14 @@ func loadMultiWorkflowScenario(t *testing.T, dirName string) Scenario {
 	}
 	return s
 }
+
+func (s Scenario) check(t *testing.T, location string) {
+	t.Helper()
+
+	if s.Name == "" {
+		t.Fatalf("fixture '%s' missing name", location)
+	}
+	if s.Event == "" {
+		t.Fatalf("fixture '%s' missing event", location)
+	}
+}

e2e/setup/agent.go

@@ -23,9 +23,6 @@ import (
 
 	"github.com/rs/zerolog/log"
 	"github.com/stretchr/testify/require"
-	"google.golang.org/grpc"
-	"google.golang.org/grpc/credentials/insecure"
-	"google.golang.org/grpc/keepalive"
 	"google.golang.org/grpc/metadata"
 
 	"go.woodpecker-ci.org/woodpecker/v3/agent"
@@ -51,9 +48,8 @@ type AgentEnv struct {
 	// name is used for logging and as the hostname label.
 	name string
 
-	// requestedOrgID is applied to the DB record by WaitForAgentRegistered
-	// so the server's GetServerLabels returns the right org-id filter.
-	// model.IDNotSet (-1) means global (default).
+	// requestOrgID is applied to the DB record by WaitForAgentRegistered.
+	// See WithOrgID. model.IDNotSet (-1) means global (default).
 	requestOrgID int64
 }
 
@@ -68,9 +64,7 @@ type agentConfig struct {
 	// They are matched against task Labels set in pipeline YAML (labels: key: value).
 	customLabels map[string]string
 
-	// orgID pins the agent to a specific organization (-1 = global).
-	// Org agents score higher than global agents for tasks in the same org,
-	// so they are always preferred by the queue when available.
+	// orgID pins the agent to a specific organization. See WithOrgID.
 	orgID int64
 }
 
@@ -82,7 +76,7 @@ func WithHostname(name string) AgentOption {
 // WithCustomLabels merges extra labels into the agent's filter set.
 // Use this to test label-based task routing, e.g.:
 //
-//	setup.StartAgent(ctx, t, addr, setup.WithCustomLabels(map[string]string{"gpu": "true"}))
+//	setup.StartAgent(t, addr, setup.WithCustomLabels(map[string]string{"gpu": "true"}))
 //
 // The pipeline YAML must set a matching label:
 //
@@ -108,6 +102,11 @@ func WithOrgID(id int64) AgentOption {
 // server at grpcAddr and returns an *AgentEnv whose AgentID is populated once
 // the agent has registered. Pass AgentOption values to configure labels, hostname,
 // or org-scoping; multiple agents can be started in the same test.
+//
+// The agent owns its own lifetime via an internal context canceled through
+// t.Cleanup. This decouples agent shutdown from the test's own context: the
+// agent keeps running until the test completes, regardless of which ctx the
+// caller's pipeline operations use.
 func StartAgent(t *testing.T, grpcAddr string, opts ...AgentOption) *AgentEnv {
 	t.Helper()
 
@@ -122,40 +121,24 @@ func StartAgent(t *testing.T, grpcAddr string, opts ...AgentOption) *AgentEnv {
 
 	env := &AgentEnv{name: cfg.hostname}
 
-	transport := grpc.WithTransportCredentials(insecure.NewCredentials())
-	keepaliveOpts := grpc.WithKeepaliveParams(keepalive.ClientParameters{
-		Time:    defaultTimeout,
-		Timeout: shortTimeout,
-	})
-
 	agentCtx, agentCancel := context.WithCancelCause(t.Context())
 	t.Cleanup(func() { agentCancel(nil) })
 
-	authConn, err := grpc.NewClient(grpcAddr, transport, keepaliveOpts)
+	agentConn, err := agent_rpc.Dial(agentCtx, agent_rpc.DialConfig{
+		ServerAddr:       grpcAddr,
+		AgentToken:       TestAgentToken,
+		AgentID:          -1,
+		Secure:           false,
+		KeepaliveTime:    defaultTimeout,
+		KeepaliveTimeout: shortTimeout,
+		AuthRefreshEvery: agentAuthRefreshEvery,
+	})
 	if err != nil {
-		t.Fatalf("StartAgent(%s): create auth gRPC connection: %v", cfg.hostname, err)
+		t.Fatalf("StartAgent(%s): dial gRPC: %v", cfg.hostname, err)
 	}
-	t.Cleanup(func() { authConn.Close() })
+	t.Cleanup(agentConn.Close)
 
-	authClient := agent_rpc.NewAuthGrpcClient(authConn, TestAgentToken, -1)
-	authInterceptor, err := agent_rpc.NewAuthInterceptor(agentCtx, authClient, agentAuthRefreshEvery)
-	if err != nil {
-		t.Fatalf("StartAgent(%s): authenticate with server: %v", cfg.hostname, err)
-	}
-
-	conn, err := grpc.NewClient(
-		grpcAddr,
-		transport,
-		keepaliveOpts,
-		grpc.WithUnaryInterceptor(authInterceptor.Unary()),
-		grpc.WithStreamInterceptor(authInterceptor.Stream()),
-	)
-	if err != nil {
-		t.Fatalf("StartAgent(%s): create main gRPC connection: %v", cfg.hostname, err)
-	}
-	t.Cleanup(func() { conn.Close() })
-
-	client := agent_rpc.NewGrpcClient(agentCtx, conn)
+	client := agent_rpc.NewGrpcClient(agentCtx, agentConn.MainConn)
 
 	grpcCtx := metadata.NewOutgoingContext(agentCtx, metadata.Pairs("hostname", cfg.hostname))
 

e2e/setup/forge.go

@@ -17,7 +17,6 @@
 package setup
 
 import (
-	"net/http"
 	"testing"
 
 	"github.com/stretchr/testify/mock"
@@ -29,13 +28,6 @@ import (
 
 // newMockForge builds a MockForge that serves the given files for any
 // config-fetch call, no-ops status reporting, and stubs all other methods safely.
-//
-// Single-workflow (len(files)==1, name ".woodpecker.yaml"): File() returns the
-// raw YAML bytes; Dir() is not called but is stubbed for safety.
-//
-// Multi-workflow (len(files)>1, names ".woodpecker/foo.yaml"): File() returns
-// empty (causing the config service to fall through to Dir()); Dir() returns
-// all files.
 func newMockForge(t *testing.T, files []*forge_types.FileMeta) *forge_mocks.MockForge {
 	t.Helper()
 	m := forge_mocks.NewMockForge(t)
@@ -44,29 +36,16 @@ func newMockForge(t *testing.T, files []*forge_types.FileMeta) *forge_mocks.Mock
 	m.On("Name").Return("mock").Maybe()
 	m.On("URL").Return("https://forge.example.test").Maybe()
 
-	if len(files) == 1 {
-		// Single-workflow: config service calls File(".woodpecker.yaml").
-		m.On(
-			"File",
-			mock.Anything, mock.Anything, mock.Anything, mock.Anything, ".woodpecker.yaml",
-		).Return(files[0].Data, nil).Maybe()
+	// we just use multi workflows
+	m.On(
+		"File",
+		mock.Anything, mock.Anything, mock.Anything, mock.Anything, mock.Anything,
+	).Return(nil, nil).Maybe()
 
-		m.On(
-			"Dir",
-			mock.Anything, mock.Anything, mock.Anything, mock.Anything, ".woodpecker",
-		).Return(files, nil).Maybe()
-	} else {
-		// Multi-workflow: config service calls Dir(".woodpecker").
-		// File() must return empty so the service falls through to Dir().
-		m.On(
-			"File",
-			mock.Anything, mock.Anything, mock.Anything, mock.Anything, ".woodpecker.yaml",
-		).Return([]byte(nil), nil).Maybe()
-		m.On(
-			"Dir",
-			mock.Anything, mock.Anything, mock.Anything, mock.Anything, ".woodpecker",
-		).Return(files, nil).Maybe()
-	}
+	m.On(
+		"Dir",
+		mock.Anything, mock.Anything, mock.Anything, mock.Anything, ".woodpecker",
+	).Return(files, nil).Maybe()
 
 	// Status reporting back to forge — no-op.
 	m.On(
@@ -82,6 +61,3 @@ func newMockForge(t *testing.T, files []*forge_types.FileMeta) *forge_mocks.Mock
 
 	return m
 }
-
-// compile-time import guard.
-var _ *http.Request

e2e/setup/server.go

@@ -25,10 +25,7 @@ import (
 	"github.com/prometheus/client_golang/prometheus"
 	"github.com/stretchr/testify/require"
 	"github.com/urfave/cli/v3"
-	"google.golang.org/grpc"
-	"google.golang.org/grpc/keepalive"
 
-	"go.woodpecker-ci.org/woodpecker/v3/rpc/proto"
 	"go.woodpecker-ci.org/woodpecker/v3/server"
 	"go.woodpecker-ci.org/woodpecker/v3/server/cache"
 	"go.woodpecker-ci.org/woodpecker/v3/server/forge"
@@ -54,7 +51,7 @@ const (
 	// TestJWTSecret is used for signing gRPC auth JWTs.
 	TestJWTSecret = "test-jwt-secret-for-integration-tests"
 
-	// TestForgeType is the forge type the mock pretends to bee.
+	// TestForgeType is the forge type the mock pretends to be.
 	TestForgeType = model.ForgeTypeGitea
 )
 
@@ -149,7 +146,10 @@ func newTestManager(s store.Store, mockForge *forge_mocks.MockForge) (services.M
 			&cli.DurationFlag{Name: "forge-timeout", Value: defaultTimeout},
 			&cli.UintFlag{Name: "forge-retry", Value: defaultRetry},
 			&cli.StringSliceFlag{Name: "environment"},
-			// Forge flags — gitea=true satisfies setupForgeService's type switch.
+			// services.NewManager reads the forge type from a cli flag and
+			// drives a type-switch in setupForgeService. We set the gitea flag
+			// to satisfy that switch; the actual forge instance is overridden
+			// below via the SetupForge hook, so the switch result is unused.
 			&cli.BoolFlag{Name: string(TestForgeType), Value: true},
 			&cli.StringFlag{Name: "forge-url", Value: "https://forge.example.test"},
 			&cli.StringSliceFlag{Name: "default-pipeline-configs", Value: constant.DefaultConfigOrder},
@@ -164,8 +164,14 @@ func newTestManager(s store.Store, mockForge *forge_mocks.MockForge) (services.M
 	return services.NewManager(cmd, s, setupForge)
 }
 
-// startGRPCServer binds to a random TCP port, registers Woodpecker's gRPC
-// services, and starts serving. Shutdown happens via t.Cleanup.
+// startGRPCServer binds to a random TCP port and serves Woodpecker's gRPC
+// services via the shared server_rpc.Serve helper. A fresh prometheus.Registry
+// is passed so subtests don't collide on metric names.
+//
+// Shutdown is synchronous: t.Cleanup cancels the serve context (triggering
+// GracefulStop inside Serve) and then blocks until Serve has returned.
+// Without this wait, the next subtest can start while the previous server's
+// goroutines are still live, which races on shared state like server.Config.
 func startGRPCServer(ctx context.Context, t *testing.T, s store.Store) string {
 	t.Helper()
 
@@ -173,44 +179,24 @@ func startGRPCServer(ctx context.Context, t *testing.T, s store.Store) string {
 	require.NoError(t, err, "listen on random port for gRPC")
 	addr := lis.Addr().String()
 
-	jwtManager := server_rpc.NewJWTManager(TestJWTSecret)
-	authorizer := server_rpc.NewAuthorizer(jwtManager)
-
-	grpcServer := grpc.NewServer(
-		grpc.StreamInterceptor(authorizer.StreamInterceptor),
-		grpc.UnaryInterceptor(authorizer.UnaryInterceptor),
-		grpc.KeepaliveEnforcementPolicy(keepalive.EnforcementPolicy{
-			MinTime: shortTimeout,
-		}),
-	)
-
-	proto.RegisterWoodpeckerServer(grpcServer, server_rpc.NewTestWoodpeckerServer(
-		server.Config.Services.Scheduler,
-		server.Config.Services.Logs,
-		s,
-		prometheus.NewRegistry(),
-	))
-	proto.RegisterWoodpeckerAuthServer(grpcServer, server_rpc.NewWoodpeckerAuthServer(
-		jwtManager,
-		TestAgentToken,
-		s,
-	))
-
+	serveCtx, cancel := context.WithCancelCause(ctx)
 	stopped := make(chan struct{})
-	grpcCtx, grpcCancel := context.WithCancelCause(ctx)
 	go func() {
-		<-grpcCtx.Done()
-		grpcServer.GracefulStop()
-		close(stopped)
-	}()
-	go func() {
-		if err := grpcServer.Serve(lis); err != nil {
-			grpcCancel(err)
-		}
+		defer close(stopped)
+		_ = server_rpc.Serve(serveCtx, server_rpc.ServeConfig{
+			Listener:         lis,
+			Store:            s,
+			Scheduler:        server.Config.Services.Scheduler,
+			Logger:           server.Config.Services.Logs,
+			JWTSecret:        TestJWTSecret,
+			AgentToken:       TestAgentToken,
+			KeepaliveMinTime: shortTimeout,
+			Registerer:       prometheus.NewRegistry(),
+		})
 	}()
 
 	t.Cleanup(func() {
-		grpcCancel(nil)
+		cancel(nil)
 		<-stopped
 	})
 	return addr

e2e/setup/wait.go

@@ -36,6 +36,21 @@ const (
 	defaultInterval = 100 * time.Millisecond
 )
 
+// pollUntil polls condition every defaultInterval until it returns true or
+// timeout is exceeded. Returns true if condition was met, false on timeout.
+// Callers are expected to handle the timeout case themselves, typically with
+// a t.Fatalf that reports the last observed state.
+func pollUntil(timeout time.Duration, condition func() bool) bool {
+	deadline := time.Now().Add(timeout)
+	for time.Now().Before(deadline) {
+		if condition() {
+			return true
+		}
+		time.Sleep(defaultInterval)
+	}
+	return condition()
+}
+
 // isTerminal returns true if the status is a final (non-running) state.
 func isTerminal(s model.StatusValue) bool {
 	switch s {
@@ -58,23 +73,20 @@ func WaitForPipeline(t *testing.T, s store.Store, pipelineID int64) *model.Pipel
 func WaitForPipelineStatus(t *testing.T, s store.Store, pipelineID int64, wantStatus model.StatusValue, timeout time.Duration) *model.Pipeline {
 	t.Helper()
 
-	deadline := time.Now().Add(timeout)
-	for time.Now().Before(deadline) {
-		p, err := s.GetPipeline(pipelineID)
+	var p *model.Pipeline
+	ok := pollUntil(timeout, func() bool {
+		var err error
+		p, err = s.GetPipeline(pipelineID)
 		require.NoError(t, err, "get pipeline %d", pipelineID)
-
 		if wantStatus != "" {
-			if p.Status == wantStatus {
-				return p
-			}
-		} else if isTerminal(p.Status) {
-			return p
+			return p.Status == wantStatus
 		}
-
-		time.Sleep(defaultInterval)
+		return isTerminal(p.Status)
+	})
+	if ok {
+		return p
 	}
 
-	p, _ := s.GetPipeline(pipelineID)
 	t.Fatalf("timeout waiting for pipeline %d: last status=%q (want %q)", pipelineID, p.Status, wantStatus)
 	return nil
 }
@@ -85,37 +97,32 @@ func WaitForPipelineStatus(t *testing.T, s store.Store, pipelineID int64, wantSt
 func WaitForAgentRegistered(t *testing.T, s store.Store, agents ...*AgentEnv) {
 	t.Helper()
 
-	deadline := time.Now().Add(shortTimeout)
-	for time.Now().Before(deadline) {
-		allFound := true
+	allRegistered := func() bool {
 		for _, env := range agents {
 			if env.AgentID == 0 {
-				allFound = false
-				break
+				return false
 			}
 			if _, err := s.AgentFind(env.AgentID); err != nil {
-				allFound = false
-				break
+				return false
 			}
 		}
-		if allFound {
-			// Apply any deferred OrgID patches.
-			for _, env := range agents {
-				if env.requestOrgID == model.IDNotSet {
-					continue
-				}
-				agent, err := s.AgentFind(env.AgentID)
-				require.NoError(t, err, "find agent %d to patch OrgID", env.AgentID)
-				agent.OrgID = env.requestOrgID
-				require.NoError(t, s.AgentUpdate(agent),
-					"patch OrgID on agent %d", env.AgentID)
-			}
-			return
-		}
-		time.Sleep(defaultInterval)
+		return true
+	}
+	if !pollUntil(shortTimeout, allRegistered) {
+		t.Fatal("timeout: not all agents registered with the server")
 	}
 
-	t.Fatal("timeout: not all agents registered with the server")
+	// Apply any deferred OrgID patches.
+	for _, env := range agents {
+		if env.requestOrgID == model.IDNotSet {
+			continue
+		}
+		agent, err := s.AgentFind(env.AgentID)
+		require.NoError(t, err, "find agent %d to patch OrgID", env.AgentID)
+		agent.OrgID = env.requestOrgID
+		require.NoError(t, s.AgentUpdate(agent),
+			"patch OrgID on agent %d", env.AgentID)
+	}
 }
 
 // WaitForStep polls the store until a named step in the given pipeline reaches
@@ -132,33 +139,29 @@ func WaitForStep(t *testing.T, s store.Store, pipeline *model.Pipeline, stepName
 func WaitForStepStatus(t *testing.T, s store.Store, pipeline *model.Pipeline, stepName string, wantState model.StatusValue, timeout time.Duration) *model.Step {
 	t.Helper()
 
-	deadline := time.Now().Add(timeout)
-	for time.Now().Before(deadline) {
+	var found *model.Step
+	ok := pollUntil(timeout, func() bool {
 		steps, err := s.StepList(pipeline.ID)
 		require.NoError(t, err, "list steps for pipeline %d", pipeline.ID)
-
 		for _, step := range steps {
 			if step.Name != stepName {
 				continue
 			}
+			found = step
 			if wantState != "" {
-				if step.State == wantState {
-					return step
-				}
-			} else if isTerminal(step.State) {
-				return step
+				return step.State == wantState
 			}
+			return isTerminal(step.State)
 		}
-		time.Sleep(defaultInterval)
+		return false
+	})
+	if ok {
+		return found
 	}
 
-	steps, _ := s.StepList(pipeline.ID)
 	var lastState model.StatusValue
-	for _, step := range steps {
-		if step.Name == stepName {
-			lastState = step.State
-			break
-		}
+	if found != nil {
+		lastState = found.State
 	}
 	if wantState != "" {
 		t.Fatalf("timeout waiting for step %q in pipeline %d to reach state %q: last state=%q",
@@ -200,13 +203,10 @@ func AssertWorkflowRanOnAgent(t *testing.T, s store.Store, pipeline *model.Pipel
 func WaitForWorkersReady(t *testing.T, q queue.Queue, minWorkers int) {
 	t.Helper()
 
-	deadline := time.Now().Add(shortTimeout)
-	for time.Now().Before(deadline) {
-		info := q.Info(context.Background())
-		if info.Stats.Workers >= minWorkers {
-			return
-		}
-		time.Sleep(defaultInterval)
+	if pollUntil(shortTimeout, func() bool {
+		return q.Info(context.Background()).Stats.Workers >= minWorkers
+	}) {
+		return
 	}
 
 	info := q.Info(context.Background())
@@ -221,21 +221,20 @@ func WaitForWorkersReady(t *testing.T, q queue.Queue, minWorkers int) {
 func WaitForStepRunning(t *testing.T, s store.Store, pipelineID int64, stepName string) {
 	t.Helper()
 
-	deadline := time.Now().Add(shortTimeout)
-	for time.Now().Before(deadline) {
+	running := pollUntil(shortTimeout, func() bool {
 		p, err := s.GetPipeline(pipelineID)
 		require.NoError(t, err, "get pipeline %d", pipelineID)
 
 		steps, err := s.StepList(p.ID)
 		require.NoError(t, err, "list steps for pipeline %d", pipelineID)
-
 		for _, step := range steps {
 			if step.Name == stepName && step.State == model.StatusRunning {
-				return
+				return true
 			}
 		}
-		time.Sleep(defaultInterval)
+		return false
+	})
+	if !running {
+		t.Fatalf("timeout waiting for step %q in pipeline %d to reach StatusRunning", stepName, pipelineID)
 	}
-
-	t.Fatalf("timeout waiting for step %q in pipeline %d to reach StatusRunning", stepName, pipelineID)
 }

server/rpc/serve.go

@@ -0,0 +1,88 @@
+// Copyright 2026 Woodpecker Authors
+//
+// Licensed under the Apache License, Version 2.0 (the "License");
+// you may not use this file except in compliance with the License.
+// You may obtain a copy of the License at
+//
+//      http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing, software
+// distributed under the License is distributed on an "AS IS" BASIS,
+// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+// See the License for the specific language governing permissions and
+// limitations under the License.
+
+package rpc
+
+import (
+	"context"
+	"fmt"
+	"net"
+	"time"
+
+	"github.com/prometheus/client_golang/prometheus"
+	"github.com/rs/zerolog/log"
+	"google.golang.org/grpc"
+	"google.golang.org/grpc/keepalive"
+
+	"go.woodpecker-ci.org/woodpecker/v3/rpc/proto"
+	"go.woodpecker-ci.org/woodpecker/v3/server/logging"
+	"go.woodpecker-ci.org/woodpecker/v3/server/scheduler"
+	"go.woodpecker-ci.org/woodpecker/v3/server/store"
+)
+
+// ServeConfig bundles everything Serve needs. Callers build this from a
+// *cli.Command (production) or with literals (tests).
+type ServeConfig struct {
+	Listener         net.Listener
+	Store            store.Store
+	Scheduler        scheduler.Scheduler
+	Logger           logging.Log
+	JWTSecret        string
+	AgentToken       string
+	KeepaliveMinTime time.Duration
+	// Registerer is where the server's prometheus metrics are registered.
+	// Pass prometheus.DefaultRegisterer in production; pass a fresh
+	// prometheus.NewRegistry() in tests to avoid duplicate-registration
+	// panics when the server is created multiple times.
+	Registerer prometheus.Registerer
+}
+
+// Serve registers Woodpecker's gRPC services on cfg.Listener and blocks
+// until ctx is canceled or Serve returns an error. GracefulStop is
+// triggered on ctx cancellation. The listener is owned by Serve — it is
+// closed when grpc.Server.Serve returns.
+func Serve(ctx context.Context, cfg ServeConfig) error {
+	jwtManager := NewJWTManager(cfg.JWTSecret)
+	authorizer := NewAuthorizer(jwtManager)
+
+	grpcServer := grpc.NewServer(
+		grpc.StreamInterceptor(authorizer.StreamInterceptor),
+		grpc.UnaryInterceptor(authorizer.UnaryInterceptor),
+		grpc.KeepaliveEnforcementPolicy(keepalive.EnforcementPolicy{
+			MinTime: cfg.KeepaliveMinTime,
+		}),
+	)
+
+	proto.RegisterWoodpeckerServer(grpcServer, NewWoodpeckerServer(
+		cfg.Scheduler, cfg.Logger, cfg.Store, cfg.Registerer,
+	))
+	proto.RegisterWoodpeckerAuthServer(grpcServer, NewWoodpeckerAuthServer(
+		jwtManager, cfg.AgentToken, cfg.Store,
+	))
+
+	grpcCtx, cancel := context.WithCancelCause(ctx)
+	defer cancel(nil)
+
+	go func() {
+		<-grpcCtx.Done()
+		log.Info().Msg("terminating grpc service gracefully")
+		grpcServer.GracefulStop()
+		log.Info().Msg("grpc service stopped")
+	}()
+
+	if err := grpcServer.Serve(cfg.Listener); err != nil {
+		return fmt.Errorf("grpc server failed: %w", err)
+	}
+	return nil
+}

server/rpc/server.go

@@ -36,33 +36,16 @@ type WoodpeckerServer struct {
 	peer RPC
 }
 
-func NewWoodpeckerServer(scheduler scheduler.Scheduler, logger logging.Log, store store.Store) proto.WoodpeckerServer {
-	pipelineTime := promauto.NewGaugeVec(prometheus.GaugeOpts{
-		Namespace: "woodpecker",
-		Name:      "pipeline_time",
-		Help:      "Pipeline time.",
-	}, []string{"repo", "branch", "status", "pipeline"})
-	pipelineCount := promauto.NewCounterVec(prometheus.CounterOpts{
-		Namespace: "woodpecker",
-		Name:      "pipeline_count",
-		Help:      "Pipeline count.",
-	}, []string{"repo", "branch", "status", "pipeline"})
-	peer := RPC{
-		store:         store,
-		scheduler:     scheduler,
-		logger:        logger,
-		pipelineTime:  pipelineTime,
-		pipelineCount: pipelineCount,
+// NewWoodpeckerServer creates a WoodpeckerServer with its metrics registered
+// into registerer. Pass prometheus.DefaultRegisterer in production; pass a
+// fresh prometheus.NewRegistry() in tests to avoid "duplicate metrics
+// collector registration" panics when the server is created multiple times.
+// A nil registerer defaults to prometheus.DefaultRegisterer.
+func NewWoodpeckerServer(scheduler scheduler.Scheduler, logger logging.Log, store store.Store, registerer prometheus.Registerer) proto.WoodpeckerServer {
+	if registerer == nil {
+		registerer = prometheus.DefaultRegisterer
 	}
-	return &WoodpeckerServer{peer: peer}
-}
-
-// NewTestWoodpeckerServer creates a WoodpeckerServer for e2e tests.
-// It is using a caller-supplied prometheus registry.
-// Use this in tests to avoid "duplicate metrics collector registration" panics when the server is created multiple times.
-// (promauto in NewWoodpeckerServer registers into the global default registry, which panics on duplicate names).
-func NewTestWoodpeckerServer(scheduler scheduler.Scheduler, logger logging.Log, store store.Store, registry *prometheus.Registry) proto.WoodpeckerServer {
-	factory := promauto.With(registry)
+	factory := promauto.With(registerer)
 	pipelineTime := factory.NewGaugeVec(prometheus.GaugeOpts{
 		Namespace: "woodpecker",
 		Name:      "pipeline_time",