perf: use single poller with semaphore-based capacity control (#822)

## Background

#819 replaced the shared `rate.Limiter` with per-worker exponential backoff counters to add jitter and adaptive polling. Before #819, the poller used:

```go
limiter := rate.NewLimiter(rate.Every(p.cfg.Runner.FetchInterval), 1)
```

This limiter was **shared across all N polling goroutines with burst=1**, effectively serializing their `FetchTask` calls — so even with `capacity=60`, the runner issued roughly one `FetchTask` per `FetchInterval` total.

#819 replaced this with independent per-worker `consecutiveEmpty` / `consecutiveErrors` counters. Each goroutine now backs off **independently**, which inadvertently removed the cross-worker serialization. With `capacity=N`, the runner now has N goroutines each polling on their own schedule — a regression from the pre-#819 baseline for any runner with `capacity > 1`.

(Thanks to @ChristopherHX for catching this in review.)

## Problem

With the post-#819 code:

- `capacity=N` maintains **N persistent polling goroutines**, each calling `FetchTask` independently
- At idle, N goroutines each wake up and send a `FetchTask` RPC per `FetchInterval`
- At full load, N goroutines **continue polling** even though no slot is available to run a new task — every one of those RPCs is wasted
- The `Shutdown()` timeout branch has a pre-existing bug: the "non-blocking check" is actually a blocking receive, so `shutdownJobs()` is never reached on timeout

## Real-World Impact: 3 Runners × capacity=60

Current production environment: 3 runners each with `capacity=60`.

| Metric | Post-#819 (current) | This PR | Reduction |
|--------|---------------------|---------|-----------|
| Polling goroutines (total) | 3 × 60 = **180** | 3 × 1 = **3** | **98.3%** (177 fewer) |
| FetchTask RPCs per poll cycle (idle) | **180** | **3** | **98.3%** |
| FetchTask RPCs per poll cycle (full load) | **180** (all wasted) | **0** (blocked on semaphore) | **100%** |
| Concurrent connections to Gitea | **180** | **3** | **98.3%** |
| Backoff state objects | 180 (per-worker) | 3 (one per runner) | Simplified |

### Idle scenario

All 180 goroutines wake up every `FetchInterval`, each sending a `FetchTask` RPC that returns empty. Server handles 180 RPCs per cycle for zero useful work. After this PR: **3 RPCs per cycle** — one per runner.

> Note: pre-#819 idle behavior was already ~3 RPCs/cycle due to the shared `rate.Limiter`. This PR restores that property while also addressing the full-load case below.

### Full-load scenario (all 180 slots occupied)

All 180 goroutines **continue polling** even though no slot is available. Every RPC is wasted. After this PR: all 3 pollers are **blocked on the semaphore** — **zero RPCs** until a task completes.

> This is a scenario neither the pre-#819 shared limiter nor the post-#819 per-worker backoff handles — both still issue `FetchTask` RPCs when no slot is free. The semaphore is the only approach of the three that ties polling to available capacity.

## Why Not Just Revert to `rate.Limiter`?

Reverting would restore the serialized behavior but is not the right long-term fix:

- **`rate.Limiter` has no concept of available capacity.** At full load it still hands out tokens and issues `FetchTask` RPCs that can't be acted on. The semaphore blocks polling entirely in that case — zero wasted RPCs.
- **It composes poorly with adaptive backoff from #819.** A shared limiter and per-worker backoff pull in different directions.
- **N goroutines serializing on a shared limiter means N-1 of them exist only to wait in line.** A single poller expresses the same behavior more directly.

The semaphore approach ties polling to capacity explicitly: `acquire slot → fetch → dispatch → release`. That invariant becomes structural rather than emergent from a rate limiter.

## Solution

Replace N polling goroutines with a **single polling loop** that uses a buffered channel as a semaphore to control concurrent task execution:

```go
// New: poller.go Poll()
sem := make(chan struct{}, p.cfg.Runner.Capacity)
for {
    select {
    case sem <- struct{}{}:       // Acquire slot (blocks at capacity)
    case <-p.pollingCtx.Done():
        return
    }
    task, ok := p.fetchTask(...)  // Single FetchTask RPC
    if !ok {
        <-sem                     // Release slot on empty response
        // backoff...
        continue
    }
    go func(t *runnerv1.Task) {   // Dispatch task
        defer func() { <-sem }()  // Release slot when done
        p.runTaskWithRecover(p.jobsCtx, t)
    }(task)
}
```

The exponential backoff and jitter from #819 are preserved — just driven by a single `workerState` instead of N per-worker states.

## Shutdown Bug Fix

Fixed a pre-existing bug in `Shutdown()` where the timeout branch could never force-cancel running jobs:

```go
// Before (BROKEN): blocking receive, shutdownJobs() never reached
_, ok := <-p.done   // blocks until p.done is closed
if !ok { return nil }
p.shutdownJobs()    // dead code when jobs are still running

// After (FIXED): proper non-blocking check
select {
case <-p.done:
    return nil
default:
}
p.shutdownJobs()    // now correctly reached on timeout
```

## Code Changes

| Area | Detail |
|------|--------|
| `Poller.runner` | `*run.Runner` → `TaskRunner` interface (enables mock-based testing) |
| `Poll()` | N goroutines → single loop with buffered-channel semaphore |
| `PollOnce()` | Inlined from removed `pollOnce()` |
| `waitBackoff()` | New helper, eliminates duplicated backoff logic |
| `resetBackoff()` | New method on `workerState`, also resets stale `lastBackoff` metric |
| `Shutdown()` | Fixed blocking receive → proper non-blocking select |
| Removed | `poll()`, `pollOnce()` private methods (-2 methods, -42 lines) |

## Test Coverage

Added `TestPoller_ConcurrencyLimitedByCapacity` which verifies:

- With `capacity=3`, at most 3 tasks execute concurrently (`maxConcurrent <= 3`)
- Tasks actually overlap in execution (`maxConcurrent >= 2`)
- `FetchTask` is never called concurrently — confirms single poller (`maxFetchConcur == 1`)
- All 6 tasks complete successfully (`totalCompleted == 6`)
- Mock runner respects context cancellation, enabling shutdown path verification

```
=== RUN   TestPoller_ConcurrencyLimitedByCapacity
--- PASS: TestPoller_ConcurrencyLimitedByCapacity (0.10s)
PASS
ok  	gitea.com/gitea/act_runner/internal/app/poll	0.59s
```

🤖 Generated with [Claude Code](https://claude.com/claude-code)

Reviewed-on: https://gitea.com/gitea/act_runner/pulls/822
Reviewed-by: silverwind <2021+silverwind@noreply.gitea.com>
Co-authored-by: Bo-Yi Wu <appleboy.tw@gmail.com>
Co-committed-by: Bo-Yi Wu <appleboy.tw@gmail.com>

.gitignore

@@ -2,7 +2,6 @@
 .env
 .runner
 coverage.txt
-/gitea-vet
 /config.yaml
 
 # Jetbrains

.golangci.yml

@@ -13,6 +13,7 @@ linters:
     - forbidigo
     - gocheckcompilerdirectives
     - gocritic
+    - goheader
     - govet
     - ineffassign
     - mirror
@@ -85,6 +86,11 @@ linters:
       enable:
         - nilness
         - unusedwrite
+    goheader:
+      values:
+        regexp:
+          HEADER: 'Copyright \d{4} The Gitea Authors\. All rights reserved\.(\nCopyright [^\n]+)*\nSPDX-License-Identifier: MIT'
+      template: '{{ HEADER }}'
   exclusions:
     generated: lax
     presets:
@@ -101,9 +107,16 @@ issues:
   max-same-issues: 0
 formatters:
   enable:
-    - gofmt
+    - gci
     - gofumpt
   settings:
+    gci:
+      custom-order: true
+      sections:
+        - standard
+        - prefix(gitea.com/gitea/act_runner)
+        - blank
+        - default
     gofumpt:
       extra-rules: true
   exclusions:

Makefile

@@ -1,6 +1,5 @@
 DIST := dist
 EXECUTABLE := act_runner
-GOFMT ?= gofumpt -l
 DIST_DIRS := $(DIST)/binaries $(DIST)/release
 GO ?= go
 SHASUM ?= shasum -a 256
@@ -12,7 +11,6 @@ GXZ_PAGAGE ?= github.com/ulikunitz/xz/cmd/gxz@v0.5.10
 LINUX_ARCHS ?= linux/amd64,linux/arm64
 DARWIN_ARCHS ?= darwin-12/amd64,darwin-12/arm64
 WINDOWS_ARCHS ?= windows/amd64
-GO_FMT_FILES := $(shell find . -type f -name "*.go" ! -name "generated.*")
 GOFILES := $(shell find . -type f -name "*.go" -o -name "go.mod" ! -name "generated.*")
 
 DOCKER_IMAGE ?= gitea/act_runner
@@ -68,19 +66,14 @@ else
 	endif
 endif
 
-GO_PACKAGES_TO_VET ?= $(filter-out gitea.com/gitea/act_runner/internal/pkg/client/mocks,$(shell $(GO) list ./...))
-
-
 TAGS ?=
 LDFLAGS ?= -X "gitea.com/gitea/act_runner/internal/pkg/ver.version=v$(RELASE_VERSION)"
 
 all: build
 
+.PHONY: fmt
 fmt:
-	@hash gofumpt > /dev/null 2>&1; if [ $$? -ne 0 ]; then \
+	$(GO) run $(GOLANGCI_LINT_PACKAGE) fmt
-		$(GO) install mvdan.cc/gofumpt@latest; \
-	fi
-	$(GOFMT) -w $(GO_FMT_FILES)
 
 .PHONY: go-check
 go-check:
@@ -93,23 +86,20 @@ go-check:
 	fi
 
 .PHONY: fmt-check
-fmt-check:
+fmt-check: fmt
-	@hash gofumpt > /dev/null 2>&1; if [ $$? -ne 0 ]; then \
+	@diff=$$(git diff --color=always); \
-		$(GO) install mvdan.cc/gofumpt@latest; \
-	fi
-	@diff=$$($(GOFMT) -d $(GO_FMT_FILES)); \
 	if [ -n "$$diff" ]; then \
 		echo "Please run 'make fmt' and commit the result:"; \
-		echo "$${diff}"; \
+		printf "%s" "$${diff}"; \
 		exit 1; \
-	fi;
+	fi
 
 .PHONY: deps-tools
 deps-tools: ## install tool dependencies
 	$(GO) install $(GOVULNCHECK_PACKAGE)
 
 .PHONY: lint
-lint: lint-go vet
+lint: lint-go
 
 .PHONY: lint-go
 lint-go: ## lint go files
@@ -139,12 +129,6 @@ tidy-check: tidy
 test: fmt-check security-check
 	@$(GO) test -race -v -cover -coverprofile coverage.txt ./... && echo "\n==>\033[32m Ok\033[m\n" || exit 1
 
-.PHONY: vet
-vet:
-	@echo "Running go vet..."
-	@$(GO) build code.gitea.io/gitea-vet
-	@$(GO) vet -vettool=gitea-vet $(GO_PACKAGES_TO_VET)
-
 install: $(GOFILES)
 	$(GO) install -v -tags '$(TAGS)' -ldflags '$(EXTLDFLAGS)-s -w $(LDFLAGS)'
 

build.go

@@ -1,11 +0,0 @@
-// Copyright 2023 The Gitea Authors. All rights reserved.
-// SPDX-License-Identifier: MIT
-
-//go:build vendor
-
-package main
-
-import (
-	// for vet
-	_ "code.gitea.io/gitea-vet"
-)

go.mod

@@ -4,7 +4,6 @@ go 1.26.0
 
 require (
 	code.gitea.io/actions-proto-go v0.4.1
-	code.gitea.io/gitea-vet v0.2.3
 	connectrpc.com/connect v1.19.1
 	github.com/avast/retry-go/v4 v4.7.0
 	github.com/docker/docker v25.0.13+incompatible
@@ -110,7 +109,6 @@ require (
 	golang.org/x/net v0.50.0 // indirect
 	golang.org/x/sync v0.19.0 // indirect
 	golang.org/x/sys v0.41.0 // indirect
-	golang.org/x/tools v0.42.0 // indirect
 	google.golang.org/genproto/googleapis/rpc v0.0.0-20240903143218-8af14fe29dc1 // indirect
 	google.golang.org/grpc v1.67.0 // indirect
 	gopkg.in/warnings.v0 v0.1.2 // indirect

go.sum

@@ -1,7 +1,5 @@
 code.gitea.io/actions-proto-go v0.4.1 h1:l0EYhjsgpUe/1VABo2eK7zcoNX2W44WOnb0MSLrKfls=
 code.gitea.io/actions-proto-go v0.4.1/go.mod h1:mn7Wkqz6JbnTOHQpot3yDeHx+O5C9EGhMEE+htvHBas=
-code.gitea.io/gitea-vet v0.2.3 h1:gdFmm6WOTM65rE8FUBTRzeQZYzXePKSSB1+r574hWwI=
-code.gitea.io/gitea-vet v0.2.3/go.mod h1:zcNbT/aJEmivCAhfmkHOlT645KNOf9W2KnkLgFjGGfE=
 connectrpc.com/connect v1.19.1 h1:R5M57z05+90EfEvCY1b7hBxDVOUl45PrtXtAV2fOC14=
 connectrpc.com/connect v1.19.1/go.mod h1:tN20fjdGlewnSFeZxLKb0xwIZ6ozc3OQs2hTXy4du9w=
 cyphar.com/go-pathrs v0.2.3 h1:0pH8gep37wB0BgaXrEaN1OtZhUMeS7VvaejSr6i822o=
@@ -224,7 +222,6 @@ github.com/xeipuuv/gojsonreference v0.0.0-20180127040603-bd5ef7bd5415 h1:EzJWgHo
 github.com/xeipuuv/gojsonreference v0.0.0-20180127040603-bd5ef7bd5415/go.mod h1:GwrjFmJcFw6At/Gs6z4yjiIwzuJ1/+UwLxMQDVQXShQ=
 github.com/xeipuuv/gojsonschema v1.2.0 h1:LhYJRs+L4fBtjZUfuSZIKGeVu0QRy8e5Xi7D17UxZ74=
 github.com/xeipuuv/gojsonschema v1.2.0/go.mod h1:anYRn/JVcOK2ZgGU+IjEV4nwlhoK5sQluxsYJ78Id3Y=
-github.com/yuin/goldmark v1.1.25/go.mod h1:3hX8gzYuyVAZsxl0MRgGTJEmQBFcNTphYh9decYSb74=
 github.com/yuin/goldmark v1.1.27/go.mod h1:3hX8gzYuyVAZsxl0MRgGTJEmQBFcNTphYh9decYSb74=
 github.com/yuin/goldmark v1.2.1/go.mod h1:3hX8gzYuyVAZsxl0MRgGTJEmQBFcNTphYh9decYSb74=
 go.etcd.io/bbolt v1.3.8/go.mod h1:N9Mkw9X8x5fupy0IKsmuqVtoGDyxsaDlbk4Rd05IAQw=
@@ -268,8 +265,6 @@ golang.org/x/exp v0.0.0-20240719175910-8a7402abbf56 h1:2dVuKD2vS7b0QIHQbpyTISPd0
 golang.org/x/exp v0.0.0-20240719175910-8a7402abbf56/go.mod h1:M4RDyNAINzryxdtnbRXRL/OHtkFuWGRjvuhBJpk2IlY=
 golang.org/x/mod v0.2.0/go.mod h1:s0Qsj1ACt9ePp/hMypM3fl4fZqREWJwdYDEqhRiZZUA=
 golang.org/x/mod v0.3.0/go.mod h1:s0Qsj1ACt9ePp/hMypM3fl4fZqREWJwdYDEqhRiZZUA=
-golang.org/x/mod v0.33.0 h1:tHFzIWbBifEmbwtGz65eaWyGiGZatSrT9prnU8DbVL8=
-golang.org/x/mod v0.33.0/go.mod h1:swjeQEj+6r7fODbD2cqrnje9PnziFuw4bmLbBZFrQ5w=
 golang.org/x/net v0.0.0-20190404232315-eb5bcb51f2a3/go.mod h1:t9HGtf8HONx5eT2rtn7q6eTqICYqUVnKs3thJo3Qplg=
 golang.org/x/net v0.0.0-20190620200207-3b0461eec859/go.mod h1:z5CRVTTTmAJ677TzLLGU+0bjPO0LkuOLi4/5GtJWs/s=
 golang.org/x/net v0.0.0-20200226121028-0de0cce0169b/go.mod h1:z5CRVTTTmAJ677TzLLGU+0bjPO0LkuOLi4/5GtJWs/s=
@@ -308,11 +303,8 @@ golang.org/x/time v0.14.0 h1:MRx4UaLrDotUKUdCIqzPC48t1Y9hANFKIRpNx+Te8PI=
 golang.org/x/time v0.14.0/go.mod h1:eL/Oa2bBBK0TkX57Fyni+NgnyQQN4LitPmob2Hjnqw4=
 golang.org/x/tools v0.0.0-20180917221912-90fa682c2a6e/go.mod h1:n7NCudcB/nEzxVGmLbDWY5pfWTLqBcC2KZ6jyYvM4mQ=
 golang.org/x/tools v0.0.0-20191119224855-298f0cb1881e/go.mod h1:b+2E5dAYhXwXZwtnZ6UAqBI28+e2cm9otk0dWdXHAEo=
-golang.org/x/tools v0.0.0-20200325010219-a49f79bcc224/go.mod h1:Sl4aGygMT6LrqrWclx+PTx3U+LnKx/seiNR+3G19Ar8=
 golang.org/x/tools v0.0.0-20200619180055-7c47624df98f/go.mod h1:EkVYQZoAsY45+roYkvgYkIh4xh/qjgUK9TdY2XT94GE=
 golang.org/x/tools v0.0.0-20210106214847-113979e3529a/go.mod h1:emZCQorbCU4vsT4fOWvOPXz4eW1wZW4PmDk9uLelYpA=
-golang.org/x/tools v0.42.0 h1:uNgphsn75Tdz5Ji2q36v/nsFSfR/9BRFvqhGBaJGd5k=
-golang.org/x/tools v0.42.0/go.mod h1:Ma6lCIwGZvHK6XtgbswSoWroEkhugApmsXyrUmBhfr0=
 golang.org/x/xerrors v0.0.0-20190717185122-a985d3407aa7/go.mod h1:I/5z698sn9Ka8TeJc9MKroUUfqBBauWjQqLJ2OPfmY0=
 golang.org/x/xerrors v0.0.0-20191011141410-1b5146add898/go.mod h1:I/5z698sn9Ka8TeJc9MKroUUfqBBauWjQqLJ2OPfmY0=
 golang.org/x/xerrors v0.0.0-20191204190536-9bdfabe68543/go.mod h1:I/5z698sn9Ka8TeJc9MKroUUfqBBauWjQqLJ2OPfmY0=

internal/app/cmd/cmd.go

@@ -8,10 +8,10 @@ import (
 	"fmt"
 	"os"
 
-	"github.com/spf13/cobra"
-
 	"gitea.com/gitea/act_runner/internal/pkg/config"
 	"gitea.com/gitea/act_runner/internal/pkg/ver"
+
+	"github.com/spf13/cobra"
 )
 
 func Execute(ctx context.Context) {

internal/app/cmd/daemon.go

@@ -16,11 +16,6 @@ import (
 	"strings"
 	"time"
 
-	"connectrpc.com/connect"
-	"github.com/mattn/go-isatty"
-	log "github.com/sirupsen/logrus"
-	"github.com/spf13/cobra"
-
 	"gitea.com/gitea/act_runner/internal/app/poll"
 	"gitea.com/gitea/act_runner/internal/app/run"
 	"gitea.com/gitea/act_runner/internal/pkg/client"
@@ -29,6 +24,11 @@ import (
 	"gitea.com/gitea/act_runner/internal/pkg/labels"
 	"gitea.com/gitea/act_runner/internal/pkg/metrics"
 	"gitea.com/gitea/act_runner/internal/pkg/ver"
+
+	"connectrpc.com/connect"
+	"github.com/mattn/go-isatty"
+	log "github.com/sirupsen/logrus"
+	"github.com/spf13/cobra"
 )
 
 func runDaemon(ctx context.Context, daemArgs *daemonArgs, configFile *string) func(cmd *cobra.Command, args []string) error {

internal/app/cmd/register.go

@@ -14,17 +14,17 @@ import (
 	"strings"
 	"time"
 
+	"gitea.com/gitea/act_runner/internal/pkg/client"
+	"gitea.com/gitea/act_runner/internal/pkg/config"
+	"gitea.com/gitea/act_runner/internal/pkg/labels"
+	"gitea.com/gitea/act_runner/internal/pkg/ver"
+
 	pingv1 "code.gitea.io/actions-proto-go/ping/v1"
 	runnerv1 "code.gitea.io/actions-proto-go/runner/v1"
 	"connectrpc.com/connect"
 	"github.com/mattn/go-isatty"
 	log "github.com/sirupsen/logrus"
 	"github.com/spf13/cobra"
-
-	"gitea.com/gitea/act_runner/internal/pkg/client"
-	"gitea.com/gitea/act_runner/internal/pkg/config"
-	"gitea.com/gitea/act_runner/internal/pkg/labels"
-	"gitea.com/gitea/act_runner/internal/pkg/ver"
 )
 
 // runRegister registers a runner to the server

internal/app/poll/poller.go

@@ -12,19 +12,24 @@ import (
 	"sync/atomic"
 	"time"
 
-	runnerv1 "code.gitea.io/actions-proto-go/runner/v1"
-	"connectrpc.com/connect"
-	log "github.com/sirupsen/logrus"
-
-	"gitea.com/gitea/act_runner/internal/app/run"
 	"gitea.com/gitea/act_runner/internal/pkg/client"
 	"gitea.com/gitea/act_runner/internal/pkg/config"
 	"gitea.com/gitea/act_runner/internal/pkg/metrics"
+
+	runnerv1 "code.gitea.io/actions-proto-go/runner/v1"
+	"connectrpc.com/connect"
+	log "github.com/sirupsen/logrus"
 )
 
+// TaskRunner abstracts task execution so the poller can be tested
+// without a real runner.
+type TaskRunner interface {
+	Run(ctx context.Context, task *runnerv1.Task) error
+}
+
 type Poller struct {
 	client       client.Client
-	runner       *run.Runner
+	runner       TaskRunner
 	cfg          *config.Config
 	tasksVersion atomic.Int64 // tasksVersion used to store the version of the last task fetched from the Gitea.
 
@@ -37,20 +42,19 @@ type Poller struct {
 	done chan struct{}
 }
 
-// workerState holds per-goroutine polling state. Backoff counters are
+// workerState holds the single poller's backoff state. Consecutive empty or
-// per-worker so that with Capacity > 1, N workers each seeing one empty
+// error responses drive exponential backoff; a successful task fetch resets
-// response don't combine into a "consecutive N empty" reading on a shared
+// both counters so the next poll fires immediately.
-// counter and trigger an unnecessarily long backoff.
 type workerState struct {
 	consecutiveEmpty  int64
 	consecutiveErrors int64
-	// lastBackoff is the last interval reported to the PollBackoffSeconds gauge
+	// lastBackoff is the last interval reported to the PollBackoffSeconds gauge;
-	// from this worker; used to suppress redundant no-op Set calls when the
+	// used to suppress redundant no-op Set calls when the backoff plateaus
-	// backoff plateaus (e.g. at FetchIntervalMax).
+	// (e.g. at FetchIntervalMax).
 	lastBackoff time.Duration
 }
 
-func New(cfg *config.Config, client client.Client, runner *run.Runner) *Poller {
+func New(cfg *config.Config, client client.Client, runner TaskRunner) *Poller {
 	pollingCtx, shutdownPolling := context.WithCancel(context.Background())
 
 	jobsCtx, shutdownJobs := context.WithCancel(context.Background())
@@ -73,22 +77,57 @@ func New(cfg *config.Config, client client.Client, runner *run.Runner) *Poller {
 }
 
 func (p *Poller) Poll() {
+	sem := make(chan struct{}, p.cfg.Runner.Capacity)
 	wg := &sync.WaitGroup{}
-	for i := 0; i < p.cfg.Runner.Capacity; i++ {
+	s := &workerState{}
-		wg.Add(1)
-		go p.poll(wg)
-	}
-	wg.Wait()
 
-	// signal that we shutdown
+	defer func() {
-	close(p.done)
+		wg.Wait()
+		close(p.done)
+	}()
+
+	for {
+		select {
+		case sem <- struct{}{}:
+		case <-p.pollingCtx.Done():
+			return
+		}
+
+		task, ok := p.fetchTask(p.pollingCtx, s)
+		if !ok {
+			<-sem
+			if !p.waitBackoff(s) {
+				return
+			}
+			continue
+		}
+
+		s.resetBackoff()
+
+		wg.Add(1)
+		go func(t *runnerv1.Task) {
+			defer wg.Done()
+			defer func() { <-sem }()
+			p.runTaskWithRecover(p.jobsCtx, t)
+		}(task)
+	}
 }
 
 func (p *Poller) PollOnce() {
-	p.pollOnce(&workerState{})
+	defer close(p.done)
-
+	s := &workerState{}
-	// signal that we're done
+	for {
-	close(p.done)
+		task, ok := p.fetchTask(p.pollingCtx, s)
+		if !ok {
+			if !p.waitBackoff(s) {
+				return
+			}
+			continue
+		}
+		s.resetBackoff()
+		p.runTaskWithRecover(p.jobsCtx, task)
+		return
+	}
 }
 
 func (p *Poller) Shutdown(ctx context.Context) error {
@@ -101,13 +140,13 @@ func (p *Poller) Shutdown(ctx context.Context) error {
 
 	// our timeout for shutting down ran out
 	case <-ctx.Done():
-		// when both the timeout fires and the graceful shutdown
+		// Both the timeout and the graceful shutdown may fire
-		// completed succsfully, this branch of the select may
+		// simultaneously. Do a non-blocking check to avoid forcing
-		// fire. Do a non-blocking check here against the graceful
+		// a shutdown when graceful already completed.
-		// shutdown status to avoid sending an error if we don't need to.
+		select {
-		_, ok := <-p.done
+		case <-p.done:
-		if !ok {
 			return nil
+		default:
 		}
 
 		// force a shutdown of all running jobs
@@ -120,18 +159,27 @@ func (p *Poller) Shutdown(ctx context.Context) error {
 	}
 }
 
-func (p *Poller) poll(wg *sync.WaitGroup) {
+func (s *workerState) resetBackoff() {
-	defer wg.Done()
+	s.consecutiveEmpty = 0
-	s := &workerState{}
+	s.consecutiveErrors = 0
-	for {
+	s.lastBackoff = 0
-		p.pollOnce(s)
+}
 
-		select {
+// waitBackoff sleeps for the current backoff interval (with jitter).
-		case <-p.pollingCtx.Done():
+// Returns false if the polling context was cancelled during the wait.
-			return
+func (p *Poller) waitBackoff(s *workerState) bool {
-		default:
+	base := p.calculateInterval(s)
-			continue
+	if base != s.lastBackoff {
-		}
+		metrics.PollBackoffSeconds.Set(base.Seconds())
+		s.lastBackoff = base
+	}
+	timer := time.NewTimer(addJitter(base))
+	select {
+	case <-timer.C:
+		return true
+	case <-p.pollingCtx.Done():
+		timer.Stop()
+		return false
 	}
 }
 
@@ -167,34 +215,6 @@ func addJitter(d time.Duration) time.Duration {
 	return d + time.Duration(jitter)
 }
 
-func (p *Poller) pollOnce(s *workerState) {
-	for {
-		task, ok := p.fetchTask(p.pollingCtx, s)
-		if !ok {
-			base := p.calculateInterval(s)
-			if base != s.lastBackoff {
-				metrics.PollBackoffSeconds.Set(base.Seconds())
-				s.lastBackoff = base
-			}
-			timer := time.NewTimer(addJitter(base))
-			select {
-			case <-timer.C:
-			case <-p.pollingCtx.Done():
-				timer.Stop()
-				return
-			}
-			continue
-		}
-
-		// Got a task — reset backoff counters for fast subsequent polling.
-		s.consecutiveEmpty = 0
-		s.consecutiveErrors = 0
-
-		p.runTaskWithRecover(p.jobsCtx, task)
-		return
-	}
-}
-
 func (p *Poller) runTaskWithRecover(ctx context.Context, task *runnerv1.Task) {
 	defer func() {
 		if r := recover(); r != nil {

internal/app/poll/poller_test.go

@@ -6,24 +6,25 @@ package poll
 import (
 	"context"
 	"errors"
+	"sync"
+	"sync/atomic"
 	"testing"
 	"time"
 
+	"gitea.com/gitea/act_runner/internal/pkg/client/mocks"
+	"gitea.com/gitea/act_runner/internal/pkg/config"
+
 	runnerv1 "code.gitea.io/actions-proto-go/runner/v1"
 	connect_go "connectrpc.com/connect"
 	"github.com/stretchr/testify/assert"
 	"github.com/stretchr/testify/mock"
 	"github.com/stretchr/testify/require"
-
-	"gitea.com/gitea/act_runner/internal/pkg/client/mocks"
-	"gitea.com/gitea/act_runner/internal/pkg/config"
 )
 
-// TestPoller_PerWorkerCounters verifies that each worker maintains its own
+// TestPoller_WorkerStateCounters verifies that workerState correctly tracks
-// backoff counters. With a shared counter, N workers each seeing one empty
+// consecutive empty responses independently per state instance, and that
-// response would inflate the counter to N and trigger an unnecessarily long
+// fetchTask increments only the relevant counter.
-// backoff. With per-worker state, each worker only sees its own count.
+func TestPoller_WorkerStateCounters(t *testing.T) {
-func TestPoller_PerWorkerCounters(t *testing.T) {
 	client := mocks.NewClient(t)
 	client.On("FetchTask", mock.Anything, mock.Anything).Return(
 		func(_ context.Context, _ *connect_go.Request[runnerv1.FetchTaskRequest]) (*connect_go.Response[runnerv1.FetchTaskResponse], error) {
@@ -77,8 +78,8 @@ func TestPoller_FetchErrorIncrementsErrorsOnly(t *testing.T) {
 	assert.Equal(t, int64(0), s.consecutiveEmpty)
 }
 
-// TestPoller_CalculateInterval verifies the per-worker exponential backoff
+// TestPoller_CalculateInterval verifies the exponential backoff math is
-// math is correctly driven by the worker's own counters.
+// correctly driven by the workerState counters.
 func TestPoller_CalculateInterval(t *testing.T) {
 	cfg, err := config.LoadDefault("")
 	require.NoError(t, err)
@@ -106,3 +107,154 @@ func TestPoller_CalculateInterval(t *testing.T) {
 		})
 	}
 }
+
+// atomicMax atomically updates target to max(target, val).
+func atomicMax(target *atomic.Int64, val int64) {
+	for {
+		old := target.Load()
+		if val <= old || target.CompareAndSwap(old, val) {
+			break
+		}
+	}
+}
+
+type mockRunner struct {
+	delay          time.Duration
+	running        atomic.Int64
+	maxConcurrent  atomic.Int64
+	totalCompleted atomic.Int64
+}
+
+func (m *mockRunner) Run(ctx context.Context, _ *runnerv1.Task) error {
+	atomicMax(&m.maxConcurrent, m.running.Add(1))
+	select {
+	case <-time.After(m.delay):
+	case <-ctx.Done():
+	}
+	m.running.Add(-1)
+	m.totalCompleted.Add(1)
+	return nil
+}
+
+// TestPoller_ConcurrencyLimitedByCapacity verifies that with capacity=3 and
+// 6 available tasks, at most 3 tasks run concurrently, and FetchTask is
+// never called concurrently (single poller).
+func TestPoller_ConcurrencyLimitedByCapacity(t *testing.T) {
+	const (
+		capacity   = 3
+		totalTasks = 6
+		taskDelay  = 50 * time.Millisecond
+	)
+
+	var (
+		tasksReturned  atomic.Int64
+		fetchConcur    atomic.Int64
+		maxFetchConcur atomic.Int64
+	)
+
+	cli := mocks.NewClient(t)
+	cli.On("FetchTask", mock.Anything, mock.Anything).Return(
+		func(_ context.Context, _ *connect_go.Request[runnerv1.FetchTaskRequest]) (*connect_go.Response[runnerv1.FetchTaskResponse], error) {
+			atomicMax(&maxFetchConcur, fetchConcur.Add(1))
+			defer fetchConcur.Add(-1)
+
+			n := tasksReturned.Add(1)
+			if n <= totalTasks {
+				return connect_go.NewResponse(&runnerv1.FetchTaskResponse{
+					Task: &runnerv1.Task{Id: n},
+				}), nil
+			}
+			return connect_go.NewResponse(&runnerv1.FetchTaskResponse{}), nil
+		},
+	)
+
+	runner := &mockRunner{delay: taskDelay}
+
+	cfg, err := config.LoadDefault("")
+	require.NoError(t, err)
+	cfg.Runner.Capacity = capacity
+	cfg.Runner.FetchInterval = 10 * time.Millisecond
+	cfg.Runner.FetchIntervalMax = 10 * time.Millisecond
+
+	poller := New(cfg, cli, runner)
+
+	var wg sync.WaitGroup
+	wg.Go(poller.Poll)
+
+	require.Eventually(t, func() bool {
+		return runner.totalCompleted.Load() >= totalTasks
+	}, 2*time.Second, 10*time.Millisecond, "all tasks should complete")
+
+	ctx, cancel := context.WithTimeout(context.Background(), time.Second)
+	defer cancel()
+	err = poller.Shutdown(ctx)
+	require.NoError(t, err)
+	wg.Wait()
+
+	assert.LessOrEqual(t, runner.maxConcurrent.Load(), int64(capacity),
+		"concurrent running tasks must not exceed capacity")
+	assert.GreaterOrEqual(t, runner.maxConcurrent.Load(), int64(2),
+		"with 6 tasks and capacity 3, at least 2 should overlap")
+	assert.Equal(t, int64(1), maxFetchConcur.Load(),
+		"FetchTask must never be called concurrently (single poller)")
+	assert.Equal(t, int64(totalTasks), runner.totalCompleted.Load(),
+		"all tasks should have been executed")
+}
+
+// TestPoller_ShutdownForcesJobsOnTimeout locks in the fix for a
+// pre-existing bug where Shutdown's timeout branch used a blocking
+// `<-p.done` receive, leaving p.shutdownJobs() unreachable. With a
+// task parked on jobsCtx and a Shutdown deadline shorter than the
+// task's natural completion, Shutdown must force-cancel via
+// shutdownJobs() and return ctx.Err() promptly — not block until the
+// task would have finished on its own.
+func TestPoller_ShutdownForcesJobsOnTimeout(t *testing.T) {
+	var served atomic.Bool
+	cli := mocks.NewClient(t)
+	cli.On("FetchTask", mock.Anything, mock.Anything).Return(
+		func(_ context.Context, _ *connect_go.Request[runnerv1.FetchTaskRequest]) (*connect_go.Response[runnerv1.FetchTaskResponse], error) {
+			if served.CompareAndSwap(false, true) {
+				return connect_go.NewResponse(&runnerv1.FetchTaskResponse{
+					Task: &runnerv1.Task{Id: 1},
+				}), nil
+			}
+			return connect_go.NewResponse(&runnerv1.FetchTaskResponse{}), nil
+		},
+	)
+
+	// delay >> Shutdown timeout: Run only returns when jobsCtx is
+	// cancelled by shutdownJobs().
+	runner := &mockRunner{delay: 30 * time.Second}
+
+	cfg, err := config.LoadDefault("")
+	require.NoError(t, err)
+	cfg.Runner.Capacity = 1
+	cfg.Runner.FetchInterval = 10 * time.Millisecond
+	cfg.Runner.FetchIntervalMax = 10 * time.Millisecond
+
+	poller := New(cfg, cli, runner)
+
+	var wg sync.WaitGroup
+	wg.Go(poller.Poll)
+
+	require.Eventually(t, func() bool {
+		return runner.running.Load() == 1
+	}, time.Second, 10*time.Millisecond, "task should start running")
+
+	ctx, cancel := context.WithTimeout(context.Background(), 50*time.Millisecond)
+	defer cancel()
+	start := time.Now()
+	err = poller.Shutdown(ctx)
+	elapsed := time.Since(start)
+
+	require.ErrorIs(t, err, context.DeadlineExceeded)
+	// With the fix, Shutdown returns shortly after the deadline once
+	// the forced job unwinds. Without the fix, the blocking <-p.done
+	// would hang for the full 30s mockRunner delay.
+	assert.Less(t, elapsed, 5*time.Second,
+		"Shutdown must not block on the parked task; shutdownJobs() must run on timeout")
+
+	wg.Wait()
+	assert.Equal(t, int64(1), runner.totalCompleted.Load(),
+		"the parked task must be cancelled and unwound")
+}

internal/app/run/runner.go

@@ -15,6 +15,13 @@ import (
 	"sync/atomic"
 	"time"
 
+	"gitea.com/gitea/act_runner/internal/pkg/client"
+	"gitea.com/gitea/act_runner/internal/pkg/config"
+	"gitea.com/gitea/act_runner/internal/pkg/labels"
+	"gitea.com/gitea/act_runner/internal/pkg/metrics"
+	"gitea.com/gitea/act_runner/internal/pkg/report"
+	"gitea.com/gitea/act_runner/internal/pkg/ver"
+
 	runnerv1 "code.gitea.io/actions-proto-go/runner/v1"
 	"connectrpc.com/connect"
 	"github.com/docker/docker/api/types/container"
@@ -23,13 +30,6 @@ import (
 	"github.com/nektos/act/pkg/model"
 	"github.com/nektos/act/pkg/runner"
 	log "github.com/sirupsen/logrus"
-
-	"gitea.com/gitea/act_runner/internal/pkg/client"
-	"gitea.com/gitea/act_runner/internal/pkg/config"
-	"gitea.com/gitea/act_runner/internal/pkg/labels"
-	"gitea.com/gitea/act_runner/internal/pkg/metrics"
-	"gitea.com/gitea/act_runner/internal/pkg/report"
-	"gitea.com/gitea/act_runner/internal/pkg/ver"
 )
 
 // Runner runs the pipeline.

internal/pkg/metrics/metrics.go

@@ -89,7 +89,7 @@ var (
 		Namespace: Namespace,
 		Subsystem: "poll",
 		Name:      "backoff_seconds",
-		Help:      "Last observed polling backoff interval. With Capacity > 1, reflects whichever worker wrote last.",
+		Help:      "Last observed polling backoff interval in seconds.",
 	})
 
 	JobsTotal = prometheus.NewCounterVec(prometheus.CounterOpts{

internal/pkg/report/reporter.go

@@ -12,16 +12,16 @@ import (
 	"sync"
 	"time"
 
+	"gitea.com/gitea/act_runner/internal/pkg/client"
+	"gitea.com/gitea/act_runner/internal/pkg/config"
+	"gitea.com/gitea/act_runner/internal/pkg/metrics"
+
 	runnerv1 "code.gitea.io/actions-proto-go/runner/v1"
 	"connectrpc.com/connect"
 	"github.com/avast/retry-go/v4"
 	log "github.com/sirupsen/logrus"
 	"google.golang.org/protobuf/proto"
 	"google.golang.org/protobuf/types/known/timestamppb"
-
-	"gitea.com/gitea/act_runner/internal/pkg/client"
-	"gitea.com/gitea/act_runner/internal/pkg/config"
-	"gitea.com/gitea/act_runner/internal/pkg/metrics"
 )
 
 type Reporter struct {

internal/pkg/report/reporter_test.go

@@ -12,6 +12,9 @@ import (
 	"testing"
 	"time"
 
+	"gitea.com/gitea/act_runner/internal/pkg/client/mocks"
+	"gitea.com/gitea/act_runner/internal/pkg/config"
+
 	runnerv1 "code.gitea.io/actions-proto-go/runner/v1"
 	connect_go "connectrpc.com/connect"
 	log "github.com/sirupsen/logrus"
@@ -20,9 +23,6 @@ import (
 	"github.com/stretchr/testify/require"
 	"google.golang.org/protobuf/types/known/structpb"
 	"google.golang.org/protobuf/types/known/timestamppb"
-
-	"gitea.com/gitea/act_runner/internal/pkg/client/mocks"
-	"gitea.com/gitea/act_runner/internal/pkg/config"
 )
 
 func TestReporter_parseLogRow(t *testing.T) {