perf: reduce runner-to-server connection load with adaptive reporting and polling (#819)

## Summary

Many teams self-host Gitea + Act Runner at scale. The current runner design causes excessive HTTP requests to the Gitea server, leading to high server load. This PR addresses three root causes: aggressive fixed-interval polling, per-task status reporting every 1 second regardless of activity, and unoptimized HTTP client configuration.

## Problem

The original architecture has these issues:

**1. Fixed 1-second reporting interval (RunDaemon)**

- Every running task calls ReportLog + ReportState every 1 second (2 HTTP requests/sec/task)
- These requests are sent even when there are no new log rows or state changes
- With 200 runners × 3 tasks each = **1,200 req/sec just for status reporting**

**2. Fixed 2-second polling interval (no backoff)**

- Idle runners poll FetchTask every 2 seconds forever, even when no jobs are queued
- No exponential backoff or jitter — all runners can synchronize after network recovery (thundering herd)
- 200 idle runners = **100 req/sec doing nothing useful**

**3. HTTP client not tuned**

- Uses http.DefaultClient with MaxIdleConnsPerHost=2, causing frequent TCP/TLS reconnects
- Creates two separate http.Client instances (one for Ping, one for Runner service) instead of sharing

**Total: ~1,300 req/sec for 200 runners with 3 tasks each**

## Solution

### Adaptive Event-Driven Log Reporting

Replace the recursive `time.AfterFunc(1s)` pattern in RunDaemon with a goroutine-based select event loop using three trigger mechanisms:

| Trigger | Default | Purpose |
|---------|---------|---------|
| `log_report_max_latency` | 3s | Guarantee even a single log line is delivered within this time |
| `log_report_interval` | 5s | Periodic sweep — steady-state cadence |
| `log_report_batch_size` | 100 rows | Immediate flush during bursty output (e.g., npm install) |

**Key design**: `log_report_max_latency` (3s) must be less than `log_report_interval` (5s) so the max-latency timer fires before the periodic ticker for single-line scenarios.

State reporting is decoupled to its own `state_report_interval` (default 5s), with immediate flush on step transitions (start/stop) via a stateNotify channel for responsive frontend UX.

Additionally:
- Skip ReportLog when `len(rows) == 0` (no pending log rows)
- Skip ReportState when `stateChanged == false && len(outputs) == 0` (nothing changed)
- Move expensive `proto.Clone` after the early-return check to avoid deep copies on no-op paths

### Polling Backoff with Jitter

Replace fixed `rate.Limiter` with adaptive exponential backoff:
- Track `consecutiveEmpty` and `consecutiveErrors` counters
- Interval doubles with each empty/error response: `base × 2^(n-1)`, capped at `fetch_interval_max` (default 60s)
- Add ±20% random jitter to prevent thundering herd
- Fetch first, sleep after ��� preserves burst=1 behavior for immediate first fetch on startup and after task completion

### HTTP Client Tuning

- Configure custom `http.Transport` with `MaxIdleConnsPerHost=10` (was 2)
- Share a single `http.Client` between PingService and RunnerService
- Add `IdleConnTimeout=90s` for clean connection lifecycle

## Load Reduction

For 200 runners × 3 tasks (70% with active log output):

| Component | Before | After | Reduction |
|-----------|--------|-------|-----------|
| Polling (idle) | 100 req/s | ~3.4 req/s | 97% |
| Log reporting | 420 req/s | ~84 req/s | 80% |
| State reporting | 126 req/s | ~25 req/s | 80% |
| **Total** | **~1,300 req/s** | **~113 req/s** | **~91%** |

## Frontend UX Impact

| Scenario | Before | After | Notes |
|----------|--------|-------|-------|
| Continuous output (npm install) | ~1s | ~5s | Periodic ticker sweep |
| Single line then silence | ~1s | ≤3s | maxLatencyTimer guarantee |
| Bursty output (100+ lines) | ~1s | <1s | Batch size immediate flush |
| Step start/stop | ~1s | <1s | stateNotify immediate flush |
| Job completion | ~1s | ~1s | Close() retry unchanged |

## New Configuration Options

All have safe defaults — existing config files need no changes:

```yaml
runner:
  fetch_interval_max: 60s        # Max backoff interval when idle
  log_report_interval: 5s        # Periodic log flush interval
  log_report_max_latency: 3s     # Max time a log row waits (must be < log_report_interval)
  log_report_batch_size: 100     # Immediate flush threshold
  state_report_interval: 5s      # State flush interval (step transitions are always immediate)
```

Config validation warns on invalid combinations:
- `fetch_interval_max < fetch_interval` → auto-corrected
- `log_report_max_latency >= log_report_interval` → warning (timer would be redundant)

## Test Plan

- [x] `go build ./...` passes
- [x] `go test ./...` passes (all existing + 3 new tests)
- [x] `golangci-lint run` — 0 issues
- [x] TestReporter_MaxLatencyTimer — verifies single log line flushed by maxLatencyTimer before logTicker
- [x] TestReporter_BatchSizeFlush — verifies batch size threshold triggers immediate flush
- [x] TestReporter_StateNotifyFlush — verifies step transition triggers immediate state flush
- [x] TestReporter_EphemeralRunnerDeletion — verifies Close/RunDaemon race safety
- [x] TestReporter_RunDaemonClose_Race — verifies concurrent Close safety

Reviewed-on: https://gitea.com/gitea/act_runner/pulls/819
Reviewed-by: Nicolas <173651+bircni@noreply.gitea.com>
Co-authored-by: Bo-Yi Wu <appleboy.tw@gmail.com>
Co-committed-by: Bo-Yi Wu <appleboy.tw@gmail.com>

go.mod

@@ -16,7 +16,7 @@ require (
 	github.com/stretchr/testify v1.11.1
 	go.yaml.in/yaml/v4 v4.0.0-rc.3
 	golang.org/x/term v0.40.0
-	golang.org/x/time v0.14.0
+	golang.org/x/time v0.14.0 // indirect
 	google.golang.org/protobuf v1.36.11
 	gopkg.in/yaml.v3 v3.0.1
 	gotest.tools/v3 v3.5.2

internal/app/poll/poller.go

@@ -7,13 +7,14 @@ import (
 	"context"
 	"errors"
 	"fmt"
+	"math/rand/v2"
 	"sync"
 	"sync/atomic"
+	"time"
 
 	runnerv1 "code.gitea.io/actions-proto-go/runner/v1"
 	"connectrpc.com/connect"
 	log "github.com/sirupsen/logrus"
-	"golang.org/x/time/rate"
 
 	"gitea.com/gitea/act_runner/internal/app/run"
 	"gitea.com/gitea/act_runner/internal/pkg/client"
@@ -35,6 +36,15 @@ type Poller struct {
 	done chan struct{}
 }
 
+// workerState holds per-goroutine polling state. Backoff counters are
+// per-worker so that with Capacity > 1, N workers each seeing one empty
+// response don't combine into a "consecutive N empty" reading on a shared
+// counter and trigger an unnecessarily long backoff.
+type workerState struct {
+	consecutiveEmpty  int64
+	consecutiveErrors int64
+}
+
 func New(cfg *config.Config, client client.Client, runner *run.Runner) *Poller {
 	pollingCtx, shutdownPolling := context.WithCancel(context.Background())
 
@@ -58,11 +68,10 @@ func New(cfg *config.Config, client client.Client, runner *run.Runner) *Poller {
 }
 
 func (p *Poller) Poll() {
-	limiter := rate.NewLimiter(rate.Every(p.cfg.Runner.FetchInterval), 1)
 	wg := &sync.WaitGroup{}
 	for i := 0; i < p.cfg.Runner.Capacity; i++ {
 		wg.Add(1)
-		go p.poll(wg, limiter)
+		go p.poll(wg)
 	}
 	wg.Wait()
 
@@ -71,9 +80,7 @@ func (p *Poller) Poll() {
 }
 
 func (p *Poller) PollOnce() {
-	limiter := rate.NewLimiter(rate.Every(p.cfg.Runner.FetchInterval), 1)
+	p.pollOnce(&workerState{})
-
-	p.pollOnce(limiter)
 
 	// signal that we're done
 	close(p.done)
@@ -108,10 +115,11 @@ func (p *Poller) Shutdown(ctx context.Context) error {
 	}
 }
 
-func (p *Poller) poll(wg *sync.WaitGroup, limiter *rate.Limiter) {
+func (p *Poller) poll(wg *sync.WaitGroup) {
 	defer wg.Done()
+	s := &workerState{}
 	for {
-		p.pollOnce(limiter)
+		p.pollOnce(s)
 
 		select {
 		case <-p.pollingCtx.Done():
@@ -122,19 +130,57 @@ func (p *Poller) poll(wg *sync.WaitGroup, limiter *rate.Limiter) {
 	}
 }
 
-func (p *Poller) pollOnce(limiter *rate.Limiter) {
+// calculateInterval returns the polling interval with exponential backoff based on
-	for {
+// consecutive empty or error responses. The interval starts at FetchInterval and
-		if err := limiter.Wait(p.pollingCtx); err != nil {
+// doubles with each consecutive empty/error, capped at FetchIntervalMax.
-			if p.pollingCtx.Err() != nil {
+func (p *Poller) calculateInterval(s *workerState) time.Duration {
-				log.WithError(err).Debug("limiter wait failed")
+	base := p.cfg.Runner.FetchInterval
+	maxInterval := p.cfg.Runner.FetchIntervalMax
+
+	n := max(s.consecutiveEmpty, s.consecutiveErrors)
+	if n <= 1 {
+		return base
 	}
+
+	// Capped exponential backoff: base * 2^(n-1), max shift=5 so multiplier <= 32
+	shift := min(n-1, 5)
+	interval := base * time.Duration(int64(1)<<shift)
+	return min(interval, maxInterval)
+}
+
+// addJitter adds +/- 20% random jitter to the given duration to avoid thundering herd.
+func addJitter(d time.Duration) time.Duration {
+	if d <= 0 {
+		return d
+	}
+	// jitter range: [-20%, +20%] of d
+	jitterRange := int64(d) * 2 / 5 // 40% total range
+	if jitterRange <= 0 {
+		return d
+	}
+	jitter := rand.Int64N(jitterRange) - jitterRange/2
+	return d + time.Duration(jitter)
+}
+
+func (p *Poller) pollOnce(s *workerState) {
+	for {
+		task, ok := p.fetchTask(p.pollingCtx, s)
+		if !ok {
+			interval := addJitter(p.calculateInterval(s))
+			timer := time.NewTimer(interval)
+			select {
+			case <-timer.C:
+			case <-p.pollingCtx.Done():
+				timer.Stop()
 				return
 			}
-		task, ok := p.fetchTask(p.pollingCtx)
-		if !ok {
 			continue
 		}
 
+		// Got a task — reset backoff counters for fast subsequent polling.
+		s.consecutiveEmpty = 0
+		s.consecutiveErrors = 0
+
 		p.runTaskWithRecover(p.jobsCtx, task)
 		return
 	}
@@ -153,7 +199,7 @@ func (p *Poller) runTaskWithRecover(ctx context.Context, task *runnerv1.Task) {
 	}
 }
 
-func (p *Poller) fetchTask(ctx context.Context) (*runnerv1.Task, bool) {
+func (p *Poller) fetchTask(ctx context.Context, s *workerState) (*runnerv1.Task, bool) {
 	reqCtx, cancel := context.WithTimeout(ctx, p.cfg.Runner.FetchTimeout)
 	defer cancel()
 
@@ -167,10 +213,15 @@ func (p *Poller) fetchTask(ctx context.Context) (*runnerv1.Task, bool) {
 	}
 	if err != nil {
 		log.WithError(err).Error("failed to fetch task")
+		s.consecutiveErrors++
 		return nil, false
 	}
 
+	// Successful response — reset error counter.
+	s.consecutiveErrors = 0
+
 	if resp == nil || resp.Msg == nil {
+		s.consecutiveEmpty++
 		return nil, false
 	}
 
@@ -179,6 +230,7 @@ func (p *Poller) fetchTask(ctx context.Context) (*runnerv1.Task, bool) {
 	}
 
 	if resp.Msg.Task == nil {
+		s.consecutiveEmpty++
 		return nil, false
 	}
 

internal/app/poll/poller_test.go

@@ -0,0 +1,108 @@
+// Copyright 2026 The Gitea Authors. All rights reserved.
+// SPDX-License-Identifier: MIT
+
+package poll
+
+import (
+	"context"
+	"errors"
+	"testing"
+	"time"
+
+	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
+// backoff counters. With a shared counter, N workers each seeing one empty
+// response would inflate the counter to N and trigger an unnecessarily long
+// backoff. With per-worker state, each worker only sees its own count.
+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) {
+			// Always return an empty response.
+			return connect_go.NewResponse(&runnerv1.FetchTaskResponse{}), nil
+		},
+	)
+
+	cfg, err := config.LoadDefault("")
+	require.NoError(t, err)
+	p := &Poller{client: client, cfg: cfg}
+
+	ctx := context.Background()
+	s1 := &workerState{}
+	s2 := &workerState{}
+
+	// Each worker independently observes one empty response.
+	_, ok := p.fetchTask(ctx, s1)
+	require.False(t, ok)
+	_, ok = p.fetchTask(ctx, s2)
+	require.False(t, ok)
+
+	assert.Equal(t, int64(1), s1.consecutiveEmpty, "worker 1 should only count its own empty response")
+	assert.Equal(t, int64(1), s2.consecutiveEmpty, "worker 2 should only count its own empty response")
+
+	// Worker 1 sees a second empty; worker 2 stays at 1.
+	_, ok = p.fetchTask(ctx, s1)
+	require.False(t, ok)
+	assert.Equal(t, int64(2), s1.consecutiveEmpty)
+	assert.Equal(t, int64(1), s2.consecutiveEmpty, "worker 2's counter must not be affected by worker 1's empty fetches")
+}
+
+// TestPoller_FetchErrorIncrementsErrorsOnly verifies that a fetch error
+// increments only the per-worker error counter, not the empty counter.
+func TestPoller_FetchErrorIncrementsErrorsOnly(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) {
+			return nil, errors.New("network unreachable")
+		},
+	)
+
+	cfg, err := config.LoadDefault("")
+	require.NoError(t, err)
+	p := &Poller{client: client, cfg: cfg}
+
+	s := &workerState{}
+	_, ok := p.fetchTask(context.Background(), s)
+	require.False(t, ok)
+	assert.Equal(t, int64(1), s.consecutiveErrors)
+	assert.Equal(t, int64(0), s.consecutiveEmpty)
+}
+
+// TestPoller_CalculateInterval verifies the per-worker exponential backoff
+// math is correctly driven by the worker's own counters.
+func TestPoller_CalculateInterval(t *testing.T) {
+	cfg, err := config.LoadDefault("")
+	require.NoError(t, err)
+	cfg.Runner.FetchInterval = 2 * time.Second
+	cfg.Runner.FetchIntervalMax = 60 * time.Second
+	p := &Poller{cfg: cfg}
+
+	cases := []struct {
+		name         string
+		empty, errs  int64
+		wantInterval time.Duration
+	}{
+		{"first poll, no backoff", 0, 0, 2 * time.Second},
+		{"single empty, still base", 1, 0, 2 * time.Second},
+		{"two empties, doubled", 2, 0, 4 * time.Second},
+		{"five empties, capped path", 5, 0, 32 * time.Second},
+		{"many empties, capped at max", 20, 0, 60 * time.Second},
+		{"errors drive backoff too", 0, 3, 8 * time.Second},
+		{"max(empty, errors) wins", 2, 4, 16 * time.Second},
+	}
+	for _, tc := range cases {
+		t.Run(tc.name, func(t *testing.T) {
+			s := &workerState{consecutiveEmpty: tc.empty, consecutiveErrors: tc.errs}
+			assert.Equal(t, tc.wantInterval, p.calculateInterval(s))
+		})
+	}
+}

internal/app/run/runner.go

@@ -98,7 +98,7 @@ func (r *Runner) Run(ctx context.Context, task *runnerv1.Task) error {
 
 	ctx, cancel := context.WithTimeout(ctx, r.cfg.Runner.Timeout)
 	defer cancel()
-	reporter := report.NewReporter(ctx, cancel, r.client, task)
+	reporter := report.NewReporter(ctx, cancel, r.client, task, r.cfg)
 	var runErr error
 	defer func() {
 		lastWords := ""

internal/pkg/client/http.go

@@ -8,6 +8,7 @@ import (
 	"crypto/tls"
 	"net/http"
 	"strings"
+	"time"
 
 	"code.gitea.io/actions-proto-go/ping/v1/pingv1connect"
 	"code.gitea.io/actions-proto-go/runner/v1/runnerv1connect"
@@ -15,16 +16,17 @@ import (
 )
 
 func getHTTPClient(endpoint string, insecure bool) *http.Client {
+	transport := &http.Transport{
+		MaxIdleConns:        10,
+		MaxIdleConnsPerHost: 10, // All requests go to one host; default is 2 which causes frequent reconnects.
+		IdleConnTimeout:     90 * time.Second,
+	}
 	if strings.HasPrefix(endpoint, "https://") && insecure {
-		return &http.Client{
+		transport.TLSClientConfig = &tls.Config{
-			Transport: &http.Transport{
-				TLSClientConfig: &tls.Config{
 			InsecureSkipVerify: true,
-				},
-			},
 		}
 	}
-	return http.DefaultClient
+	return &http.Client{Transport: transport}
 }
 
 // New returns a new runner client.
@@ -47,14 +49,15 @@ func New(endpoint string, insecure bool, uuid, token, version string, opts ...co
 		}
 	})))
 
+	httpClient := getHTTPClient(endpoint, insecure)
 	return &HTTPClient{
 		PingServiceClient: pingv1connect.NewPingServiceClient(
-			getHTTPClient(endpoint, insecure),
+			httpClient,
 			baseURL,
 			opts...,
 		),
 		RunnerServiceClient: runnerv1connect.NewRunnerServiceClient(
-			getHTTPClient(endpoint, insecure),
+			httpClient,
 			baseURL,
 			opts...,
 		),

internal/pkg/config/config.example.yaml

@@ -32,6 +32,24 @@ runner:
   fetch_timeout: 5s
   # The interval for fetching the job from the Gitea instance.
   fetch_interval: 2s
+  # The maximum interval for fetching the job from the Gitea instance.
+  # The runner uses exponential backoff when idle, increasing the interval up to this maximum.
+  # Set to 0 or same as fetch_interval to disable backoff.
+  fetch_interval_max: 60s
+  # The base interval for periodic log flush to the Gitea instance.
+  # Logs may be sent earlier if the buffer reaches log_report_batch_size
+  # or if log_report_max_latency expires after the first buffered row.
+  log_report_interval: 5s
+  # The maximum time a log row can wait before being sent.
+  # This ensures even a single log line appears on the frontend within this duration.
+  # Must be less than log_report_interval to have any effect.
+  log_report_max_latency: 3s
+  # Flush logs immediately when the buffer reaches this many rows.
+  # This ensures bursty output (e.g., npm install) is delivered promptly.
+  log_report_batch_size: 100
+  # The interval for reporting task state (step status, timing) to the Gitea instance.
+  # State is also reported immediately on step transitions (start/stop).
+  state_report_interval: 5s
   # The github_mirror of a runner is used to specify the mirror address of the github that pulls the action repository.
   # It works when something like `uses: actions/checkout@v4` is used and DEFAULT_ACTIONS_URL is set to github,
   # and github_mirror is not empty. In this case,

internal/pkg/config/config.go

@@ -31,6 +31,11 @@ type Runner struct {
 	Insecure            bool              `yaml:"insecure"`               // Insecure indicates whether the runner operates in an insecure mode.
 	FetchTimeout        time.Duration     `yaml:"fetch_timeout"`          // FetchTimeout specifies the timeout duration for fetching resources.
 	FetchInterval       time.Duration     `yaml:"fetch_interval"`         // FetchInterval specifies the interval duration for fetching resources.
+	FetchIntervalMax    time.Duration     `yaml:"fetch_interval_max"`     // FetchIntervalMax specifies the maximum backoff interval when idle.
+	LogReportInterval   time.Duration     `yaml:"log_report_interval"`    // LogReportInterval specifies the base interval for periodic log flush.
+	LogReportMaxLatency time.Duration     `yaml:"log_report_max_latency"` // LogReportMaxLatency specifies the max time a log row can wait before being sent.
+	LogReportBatchSize  int               `yaml:"log_report_batch_size"`  // LogReportBatchSize triggers immediate log flush when buffer reaches this size.
+	StateReportInterval time.Duration     `yaml:"state_report_interval"`  // StateReportInterval specifies the interval for state reporting.
 	Labels              []string          `yaml:"labels"`                 // Labels specify the labels of the runner. Labels are declared on each startup
 	GithubMirror        string            `yaml:"github_mirror"`          // GithubMirror defines what mirrors should be used when using github
 }
@@ -137,6 +142,32 @@ func LoadDefault(file string) (*Config, error) {
 	if cfg.Runner.FetchInterval <= 0 {
 		cfg.Runner.FetchInterval = 2 * time.Second
 	}
+	if cfg.Runner.FetchIntervalMax <= 0 {
+		cfg.Runner.FetchIntervalMax = 60 * time.Second
+	}
+	if cfg.Runner.LogReportInterval <= 0 {
+		cfg.Runner.LogReportInterval = 5 * time.Second
+	}
+	if cfg.Runner.LogReportMaxLatency <= 0 {
+		cfg.Runner.LogReportMaxLatency = 3 * time.Second
+	}
+	if cfg.Runner.LogReportBatchSize <= 0 {
+		cfg.Runner.LogReportBatchSize = 100
+	}
+	if cfg.Runner.StateReportInterval <= 0 {
+		cfg.Runner.StateReportInterval = 5 * time.Second
+	}
+
+	// Validate and fix invalid config combinations to prevent confusing behavior.
+	if cfg.Runner.FetchIntervalMax < cfg.Runner.FetchInterval {
+		log.Warnf("fetch_interval_max (%v) is less than fetch_interval (%v), setting fetch_interval_max to fetch_interval",
+			cfg.Runner.FetchIntervalMax, cfg.Runner.FetchInterval)
+		cfg.Runner.FetchIntervalMax = cfg.Runner.FetchInterval
+	}
+	if cfg.Runner.LogReportMaxLatency >= cfg.Runner.LogReportInterval {
+		log.Warnf("log_report_max_latency (%v) >= log_report_interval (%v), the max-latency timer will never fire before the periodic ticker; consider lowering log_report_max_latency",
+			cfg.Runner.LogReportMaxLatency, cfg.Runner.LogReportInterval)
+	}
 
 	// although `container.network_mode` will be deprecated, but we have to be compatible with it for now.
 	if cfg.Container.NetworkMode != "" && cfg.Container.Network == "" {

internal/pkg/report/reporter.go

@@ -20,6 +20,7 @@ import (
 	"google.golang.org/protobuf/types/known/timestamppb"
 
 	"gitea.com/gitea/act_runner/internal/pkg/client"
+	"gitea.com/gitea/act_runner/internal/pkg/config"
 )
 
 type Reporter struct {
@@ -36,15 +37,26 @@ type Reporter struct {
 	oldnew      []string
 
 	state        *runnerv1.TaskState
+	stateChanged bool
 	stateMu      sync.RWMutex
 	outputs      sync.Map
 	daemon       chan struct{}
 
+	// Adaptive batching control
+	logReportInterval   time.Duration
+	logReportMaxLatency time.Duration
+	logBatchSize        int
+	stateReportInterval time.Duration
+
+	// Event notification channels (non-blocking, buffered 1)
+	logNotify   chan struct{} // signal: new log rows arrived
+	stateNotify chan struct{} // signal: step transition (start/stop)
+
 	debugOutputEnabled  bool
 	stopCommandEndToken string
 }
 
-func NewReporter(ctx context.Context, cancel context.CancelFunc, client client.Client, task *runnerv1.Task) *Reporter {
+func NewReporter(ctx context.Context, cancel context.CancelFunc, client client.Client, task *runnerv1.Task, cfg *config.Config) *Reporter {
 	var oldnew []string
 	if v := task.Context.Fields["token"].GetStringValue(); v != "" {
 		oldnew = append(oldnew, v, "***")
@@ -62,6 +74,12 @@ func NewReporter(ctx context.Context, cancel context.CancelFunc, client client.C
 		client:              client,
 		oldnew:              oldnew,
 		logReplacer:         strings.NewReplacer(oldnew...),
+		logReportInterval:   cfg.Runner.LogReportInterval,
+		logReportMaxLatency: cfg.Runner.LogReportMaxLatency,
+		logBatchSize:        cfg.Runner.LogReportBatchSize,
+		stateReportInterval: cfg.Runner.StateReportInterval,
+		logNotify:           make(chan struct{}, 1),
+		stateNotify:         make(chan struct{}, 1),
 		state: &runnerv1.TaskState{
 			Id: task.Id,
 		},
@@ -108,11 +126,42 @@ func isJobStepEntry(entry *log.Entry) bool {
 	return true
 }
 
-func (r *Reporter) Fire(entry *log.Entry) error {
+// notifyLog sends a non-blocking signal that new log rows are available.
-	r.stateMu.Lock()
+func (r *Reporter) notifyLog() {
-	defer r.stateMu.Unlock()
+	select {
+	case r.logNotify <- struct{}{}:
+	default:
+	}
+}
 
+// notifyState sends a non-blocking signal that a UX-critical state change occurred (step start/stop, job result).
+func (r *Reporter) notifyState() {
+	select {
+	case r.stateNotify <- struct{}{}:
+	default:
+	}
+}
+
+// unlockAndNotify releases stateMu and sends channel notifications.
+// Must be called with stateMu held.
+func (r *Reporter) unlockAndNotify(urgentState bool) {
+	r.stateMu.Unlock()
+	r.notifyLog()
+	if urgentState {
+		r.notifyState()
+	}
+}
+
+func (r *Reporter) Fire(entry *log.Entry) error {
+	urgentState := false
+
+	r.stateMu.Lock()
+
+	r.stateChanged = true
+
+	if log.IsLevelEnabled(log.TraceLevel) {
 		log.WithFields(entry.Data).Trace(entry.Message)
+	}
 
 	timestamp := entry.Time
 	if r.state.StartedAt == nil {
@@ -135,11 +184,13 @@ func (r *Reporter) Fire(entry *log.Entry) error {
 						}
 					}
 				}
+				urgentState = true
 			}
 		}
 		if !r.duringSteps() {
 			r.logRows = appendIfNotNil(r.logRows, r.parseLogRow(entry))
 		}
+		r.unlockAndNotify(urgentState)
 		return nil
 	}
 
@@ -153,11 +204,13 @@ func (r *Reporter) Fire(entry *log.Entry) error {
 		if !r.duringSteps() {
 			r.logRows = appendIfNotNil(r.logRows, r.parseLogRow(entry))
 		}
+		r.unlockAndNotify(false)
 		return nil
 	}
 
 	if step.StartedAt == nil {
 		step.StartedAt = timestamppb.New(timestamp)
+		urgentState = true
 	}
 
 	// Force reporting log errors as raw output to prevent silent failures
@@ -185,26 +238,91 @@ func (r *Reporter) Fire(entry *log.Entry) error {
 			}
 			step.Result = stepResult
 			step.StoppedAt = timestamppb.New(timestamp)
+			urgentState = true
 		}
 	}
 
+	r.unlockAndNotify(urgentState)
 	return nil
 }
 
 func (r *Reporter) RunDaemon() {
+	go r.runDaemonLoop()
+}
+
+func (r *Reporter) stopLatencyTimer(active *bool, timer *time.Timer) {
+	if *active {
+		if !timer.Stop() {
+			select {
+			case <-timer.C:
+			default:
+			}
+		}
+		*active = false
+	}
+}
+
+func (r *Reporter) runDaemonLoop() {
+	logTicker := time.NewTicker(r.logReportInterval)
+	stateTicker := time.NewTicker(r.stateReportInterval)
+
+	// maxLatencyTimer ensures the first buffered log row is sent within logReportMaxLatency.
+	// Start inactive — it is armed when the first log row arrives in an empty buffer.
+	maxLatencyTimer := time.NewTimer(0)
+	if !maxLatencyTimer.Stop() {
+		<-maxLatencyTimer.C
+	}
+	maxLatencyActive := false
+
+	defer logTicker.Stop()
+	defer stateTicker.Stop()
+	defer maxLatencyTimer.Stop()
+
+	for {
+		select {
+		case <-logTicker.C:
+			_ = r.ReportLog(false)
+			r.stopLatencyTimer(&maxLatencyActive, maxLatencyTimer)
+
+		case <-stateTicker.C:
+			_ = r.ReportState(false)
+
+		case <-r.logNotify:
 			r.stateMu.RLock()
-	closed := r.closed
+			n := len(r.logRows)
 			r.stateMu.RUnlock()
-	if closed || r.ctx.Err() != nil {
+
-		// Acknowledge close
+			if n >= r.logBatchSize {
+				_ = r.ReportLog(false)
+				r.stopLatencyTimer(&maxLatencyActive, maxLatencyTimer)
+			} else if !maxLatencyActive && n > 0 {
+				maxLatencyTimer.Reset(r.logReportMaxLatency)
+				maxLatencyActive = true
+			}
+
+		case <-r.stateNotify:
+			// Step transition or job result — flush both immediately for frontend UX.
+			_ = r.ReportLog(false)
+			_ = r.ReportState(false)
+			r.stopLatencyTimer(&maxLatencyActive, maxLatencyTimer)
+
+		case <-maxLatencyTimer.C:
+			maxLatencyActive = false
+			_ = r.ReportLog(false)
+
+		case <-r.ctx.Done():
 			close(r.daemon)
 			return
 		}
 
-	_ = r.ReportLog(false)
+		r.stateMu.RLock()
-	_ = r.ReportState(false)
+		closed := r.closed
-
+		r.stateMu.RUnlock()
-	time.AfterFunc(time.Second, r.RunDaemon)
+		if closed {
+			close(r.daemon)
+			return
+		}
+	}
 }
 
 func (r *Reporter) Logf(format string, a ...any) {
@@ -268,6 +386,10 @@ func (r *Reporter) Close(lastWords string) error {
 		})
 	}
 	r.stateMu.Unlock()
+
+	// Wake up the daemon loop so it detects closed promptly.
+	r.notifyLog()
+
 	// Wait for Acknowledge
 	select {
 	case <-r.daemon:
@@ -295,6 +417,10 @@ func (r *Reporter) ReportLog(noMore bool) error {
 	rows := r.logRows
 	r.stateMu.RUnlock()
 
+	if !noMore && len(rows) == 0 {
+		return nil
+	}
+
 	resp, err := r.client.UpdateLog(r.ctx, connect.NewRequest(&runnerv1.UpdateLogRequest{
 		TaskId: r.state.Id,
 		Index:  int64(r.logOffset),
@@ -329,15 +455,7 @@ func (r *Reporter) ReportState(reportResult bool) error {
 	r.clientM.Lock()
 	defer r.clientM.Unlock()
 
-	r.stateMu.RLock()
+	// Build the outputs map first (single Range pass instead of two).
-	state := proto.Clone(r.state).(*runnerv1.TaskState)
-	r.stateMu.RUnlock()
-
-	// Only report result from Close to reliable sent logs
-	if !reportResult {
-		state.Result = runnerv1.Result_RESULT_UNSPECIFIED
-	}
-
 	outputs := make(map[string]string)
 	r.outputs.Range(func(k, v any) bool {
 		if val, ok := v.(string); ok {
@@ -346,11 +464,29 @@ func (r *Reporter) ReportState(reportResult bool) error {
 		return true
 	})
 
+	// Consume stateChanged atomically with the snapshot; restored on error
+	// below so a concurrent Fire() during UpdateTask isn't silently lost.
+	r.stateMu.Lock()
+	if !reportResult && !r.stateChanged && len(outputs) == 0 {
+		r.stateMu.Unlock()
+		return nil
+	}
+	state := proto.Clone(r.state).(*runnerv1.TaskState)
+	r.stateChanged = false
+	r.stateMu.Unlock()
+
+	if !reportResult {
+		state.Result = runnerv1.Result_RESULT_UNSPECIFIED
+	}
+
 	resp, err := r.client.UpdateTask(r.ctx, connect.NewRequest(&runnerv1.UpdateTaskRequest{
 		State:   state,
 		Outputs: outputs,
 	}))
 	if err != nil {
+		r.stateMu.Lock()
+		r.stateChanged = true
+		r.stateMu.Unlock()
 		return err
 	}
 

internal/pkg/report/reporter_test.go

@@ -6,8 +6,9 @@ package report
 import (
 	"context"
 	"errors"
+	"fmt"
 	"strings"
-	"sync"
+	"sync/atomic"
 	"testing"
 	"time"
 
@@ -21,6 +22,7 @@ import (
 	"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) {
@@ -175,9 +177,10 @@ func TestReporter_Fire(t *testing.T) {
 		ctx, cancel := context.WithCancel(context.Background())
 		taskCtx, err := structpb.NewStruct(map[string]any{})
 		require.NoError(t, err)
+		cfg, _ := config.LoadDefault("")
 		reporter := NewReporter(ctx, cancel, client, &runnerv1.Task{
 			Context: taskCtx,
-		})
+		}, cfg)
 		reporter.RunDaemon()
 		defer func() {
 			require.NoError(t, reporter.Close(""))
@@ -252,7 +255,8 @@ func TestReporter_EphemeralRunnerDeletion(t *testing.T) {
 	defer cancel()
 	taskCtx, err := structpb.NewStruct(map[string]any{})
 	require.NoError(t, err)
-	reporter := NewReporter(ctx, cancel, client, &runnerv1.Task{Context: taskCtx})
+	cfg, _ := config.LoadDefault("")
+	reporter := NewReporter(ctx, cancel, client, &runnerv1.Task{Context: taskCtx}, cfg)
 	reporter.ResetSteps(1)
 
 	// Fire a log entry to create pending data
@@ -315,23 +319,281 @@ func TestReporter_RunDaemonClose_Race(t *testing.T) {
 	ctx, cancel := context.WithCancel(context.Background())
 	taskCtx, err := structpb.NewStruct(map[string]any{})
 	require.NoError(t, err)
+	cfg, _ := config.LoadDefault("")
 	reporter := NewReporter(ctx, cancel, client, &runnerv1.Task{
 		Context: taskCtx,
-	})
+	}, cfg)
 	reporter.ResetSteps(1)
 
-	// Start the daemon loop in a separate goroutine.
+	// Start the daemon loop — RunDaemon spawns a goroutine internally.
-	// RunDaemon reads r.closed and reschedules itself via time.AfterFunc.
-	var wg sync.WaitGroup
-	wg.Go(func() {
 	reporter.RunDaemon()
-	})
 
-	// Close concurrently — this races with RunDaemon on r.closed.
+	// Close concurrently — this races with the daemon goroutine on r.closed.
 	require.NoError(t, reporter.Close(""))
 
-	// Cancel context so pending AfterFunc callbacks exit quickly.
+	// Cancel context so the daemon goroutine exits cleanly.
 	cancel()
-	wg.Wait()
+}
-	time.Sleep(2 * time.Second)
+
+// TestReporter_MaxLatencyTimer verifies that the maxLatencyTimer flushes a
+// single buffered log row before the periodic logTicker fires.
+//
+// Setup: logReportInterval=10s (effectively never), maxLatency=100ms.
+// Fire one log line, then assert UpdateLog is called within 500ms.
+func TestReporter_MaxLatencyTimer(t *testing.T) {
+	var updateLogCalls atomic.Int64
+
+	client := mocks.NewClient(t)
+	client.On("UpdateLog", mock.Anything, mock.Anything).Return(
+		func(_ context.Context, req *connect_go.Request[runnerv1.UpdateLogRequest]) (*connect_go.Response[runnerv1.UpdateLogResponse], error) {
+			updateLogCalls.Add(1)
+			return connect_go.NewResponse(&runnerv1.UpdateLogResponse{
+				AckIndex: req.Msg.Index + int64(len(req.Msg.Rows)),
+			}), nil
+		},
+	)
+	client.On("UpdateTask", mock.Anything, mock.Anything).Maybe().Return(
+		func(_ context.Context, _ *connect_go.Request[runnerv1.UpdateTaskRequest]) (*connect_go.Response[runnerv1.UpdateTaskResponse], error) {
+			return connect_go.NewResponse(&runnerv1.UpdateTaskResponse{}), nil
+		},
+	)
+
+	ctx, cancel := context.WithCancel(context.Background())
+	defer cancel()
+	taskCtx, err := structpb.NewStruct(map[string]any{})
+	require.NoError(t, err)
+
+	// Custom config: logTicker=10s (won't fire during test), maxLatency=100ms
+	cfg, _ := config.LoadDefault("")
+	cfg.Runner.LogReportInterval = 10 * time.Second
+	cfg.Runner.LogReportMaxLatency = 100 * time.Millisecond
+	cfg.Runner.LogReportBatchSize = 1000 // won't trigger batch flush
+
+	reporter := NewReporter(ctx, cancel, client, &runnerv1.Task{Context: taskCtx}, cfg)
+	reporter.ResetSteps(1)
+	reporter.RunDaemon()
+	defer func() {
+		_ = reporter.Close("")
+	}()
+
+	// Fire a single log line — not enough to trigger batch flush
+	require.NoError(t, reporter.Fire(&log.Entry{
+		Message: "single log line",
+		Data:    log.Fields{"stage": "Main", "stepNumber": 0, "raw_output": true},
+	}))
+
+	// maxLatencyTimer should flush within ~100ms. Wait up to 500ms.
+	assert.Eventually(t, func() bool {
+		return updateLogCalls.Load() > 0
+	}, 500*time.Millisecond, 10*time.Millisecond,
+		"maxLatencyTimer should have flushed the log before logTicker (10s)")
+}
+
+// TestReporter_BatchSizeFlush verifies that reaching logBatchSize triggers
+// an immediate log flush without waiting for any timer.
+func TestReporter_BatchSizeFlush(t *testing.T) {
+	var updateLogCalls atomic.Int64
+
+	client := mocks.NewClient(t)
+	client.On("UpdateLog", mock.Anything, mock.Anything).Return(
+		func(_ context.Context, req *connect_go.Request[runnerv1.UpdateLogRequest]) (*connect_go.Response[runnerv1.UpdateLogResponse], error) {
+			updateLogCalls.Add(1)
+			return connect_go.NewResponse(&runnerv1.UpdateLogResponse{
+				AckIndex: req.Msg.Index + int64(len(req.Msg.Rows)),
+			}), nil
+		},
+	)
+	client.On("UpdateTask", mock.Anything, mock.Anything).Maybe().Return(
+		func(_ context.Context, _ *connect_go.Request[runnerv1.UpdateTaskRequest]) (*connect_go.Response[runnerv1.UpdateTaskResponse], error) {
+			return connect_go.NewResponse(&runnerv1.UpdateTaskResponse{}), nil
+		},
+	)
+
+	ctx, cancel := context.WithCancel(context.Background())
+	defer cancel()
+	taskCtx, err := structpb.NewStruct(map[string]any{})
+	require.NoError(t, err)
+
+	// Custom config: large timers, small batch size
+	cfg, _ := config.LoadDefault("")
+	cfg.Runner.LogReportInterval = 10 * time.Second
+	cfg.Runner.LogReportMaxLatency = 10 * time.Second
+	cfg.Runner.LogReportBatchSize = 5
+
+	reporter := NewReporter(ctx, cancel, client, &runnerv1.Task{Context: taskCtx}, cfg)
+	reporter.ResetSteps(1)
+	reporter.RunDaemon()
+	defer func() {
+		_ = reporter.Close("")
+	}()
+
+	// Fire exactly batchSize log lines
+	for i := range 5 {
+		require.NoError(t, reporter.Fire(&log.Entry{
+			Message: fmt.Sprintf("log line %d", i),
+			Data:    log.Fields{"stage": "Main", "stepNumber": 0, "raw_output": true},
+		}))
+	}
+
+	// Batch threshold should trigger immediate flush
+	assert.Eventually(t, func() bool {
+		return updateLogCalls.Load() > 0
+	}, 500*time.Millisecond, 10*time.Millisecond,
+		"batch size threshold should have triggered immediate flush")
+}
+
+// TestReporter_StateChangedNotLostDuringReport asserts that a Fire() arriving
+// mid-UpdateTask re-dirties the flag so the change is picked up by the next report.
+func TestReporter_StateChangedNotLostDuringReport(t *testing.T) {
+	var updateTaskCalls atomic.Int64
+	inFlight := make(chan struct{})
+	release := make(chan struct{})
+
+	client := mocks.NewClient(t)
+	client.On("UpdateTask", mock.Anything, mock.Anything).Return(
+		func(_ context.Context, _ *connect_go.Request[runnerv1.UpdateTaskRequest]) (*connect_go.Response[runnerv1.UpdateTaskResponse], error) {
+			n := updateTaskCalls.Add(1)
+			if n == 1 {
+				// Signal that the first UpdateTask is in flight, then block until released.
+				close(inFlight)
+				<-release
+			}
+			return connect_go.NewResponse(&runnerv1.UpdateTaskResponse{}), nil
+		},
+	)
+
+	ctx, cancel := context.WithCancel(context.Background())
+	defer cancel()
+	taskCtx, err := structpb.NewStruct(map[string]any{})
+	require.NoError(t, err)
+	cfg, _ := config.LoadDefault("")
+	reporter := NewReporter(ctx, cancel, client, &runnerv1.Task{Context: taskCtx}, cfg)
+	reporter.ResetSteps(2)
+
+	// Mark stateChanged=true so the first ReportState proceeds to UpdateTask.
+	reporter.stateMu.Lock()
+	reporter.stateChanged = true
+	reporter.stateMu.Unlock()
+
+	// Kick off the first ReportState in a goroutine — it will block in UpdateTask.
+	done := make(chan error, 1)
+	go func() {
+		done <- reporter.ReportState(false)
+	}()
+
+	// Wait until UpdateTask is in flight (snapshot taken, flag consumed).
+	<-inFlight
+
+	// Concurrent Fire() modifies state — must re-flip stateChanged so the
+	// change is not lost when the in-flight ReportState finishes.
+	require.NoError(t, reporter.Fire(&log.Entry{
+		Message: "step starts",
+		Data:    log.Fields{"stage": "Main", "stepNumber": 1, "raw_output": true},
+	}))
+
+	// Release the in-flight UpdateTask and wait for it to return.
+	close(release)
+	require.NoError(t, <-done)
+
+	// stateChanged must still be true so the next ReportState picks up the
+	// concurrent Fire()'s change instead of skipping via the early-return path.
+	reporter.stateMu.RLock()
+	changed := reporter.stateChanged
+	reporter.stateMu.RUnlock()
+	assert.True(t, changed, "stateChanged must remain true after a concurrent Fire() during in-flight ReportState")
+
+	// And the next ReportState must actually send a second UpdateTask.
+	require.NoError(t, reporter.ReportState(false))
+	assert.Equal(t, int64(2), updateTaskCalls.Load(), "concurrent Fire() change must trigger a second UpdateTask, not be silently lost")
+}
+
+// TestReporter_StateChangedRestoredOnError verifies that when UpdateTask fails,
+// the dirty flag is restored so the snapshotted change isn't silently lost.
+func TestReporter_StateChangedRestoredOnError(t *testing.T) {
+	var updateTaskCalls atomic.Int64
+
+	client := mocks.NewClient(t)
+	client.On("UpdateTask", mock.Anything, mock.Anything).Return(
+		func(_ context.Context, _ *connect_go.Request[runnerv1.UpdateTaskRequest]) (*connect_go.Response[runnerv1.UpdateTaskResponse], error) {
+			n := updateTaskCalls.Add(1)
+			if n == 1 {
+				return nil, errors.New("transient network error")
+			}
+			return connect_go.NewResponse(&runnerv1.UpdateTaskResponse{}), nil
+		},
+	)
+
+	ctx, cancel := context.WithCancel(context.Background())
+	defer cancel()
+	taskCtx, err := structpb.NewStruct(map[string]any{})
+	require.NoError(t, err)
+	cfg, _ := config.LoadDefault("")
+	reporter := NewReporter(ctx, cancel, client, &runnerv1.Task{Context: taskCtx}, cfg)
+	reporter.ResetSteps(1)
+
+	reporter.stateMu.Lock()
+	reporter.stateChanged = true
+	reporter.stateMu.Unlock()
+
+	// First ReportState fails — flag must be restored to true.
+	require.Error(t, reporter.ReportState(false))
+
+	reporter.stateMu.RLock()
+	changed := reporter.stateChanged
+	reporter.stateMu.RUnlock()
+	assert.True(t, changed, "stateChanged must be restored to true after UpdateTask error so the change is retried")
+
+	// The next ReportState should still issue a request because the flag was restored.
+	require.NoError(t, reporter.ReportState(false))
+	assert.Equal(t, int64(2), updateTaskCalls.Load())
+}
+
+// TestReporter_StateNotifyFlush verifies that step transitions trigger
+// an immediate state flush via the stateNotify channel.
+func TestReporter_StateNotifyFlush(t *testing.T) {
+	var updateTaskCalls atomic.Int64
+
+	client := mocks.NewClient(t)
+	client.On("UpdateLog", mock.Anything, mock.Anything).Maybe().Return(
+		func(_ context.Context, req *connect_go.Request[runnerv1.UpdateLogRequest]) (*connect_go.Response[runnerv1.UpdateLogResponse], error) {
+			return connect_go.NewResponse(&runnerv1.UpdateLogResponse{
+				AckIndex: req.Msg.Index + int64(len(req.Msg.Rows)),
+			}), nil
+		},
+	)
+	client.On("UpdateTask", mock.Anything, mock.Anything).Return(
+		func(_ context.Context, _ *connect_go.Request[runnerv1.UpdateTaskRequest]) (*connect_go.Response[runnerv1.UpdateTaskResponse], error) {
+			updateTaskCalls.Add(1)
+			return connect_go.NewResponse(&runnerv1.UpdateTaskResponse{}), nil
+		},
+	)
+
+	ctx, cancel := context.WithCancel(context.Background())
+	defer cancel()
+	taskCtx, err := structpb.NewStruct(map[string]any{})
+	require.NoError(t, err)
+
+	// Custom config: large state interval so only stateNotify can trigger
+	cfg, _ := config.LoadDefault("")
+	cfg.Runner.StateReportInterval = 10 * time.Second
+	cfg.Runner.LogReportInterval = 10 * time.Second
+
+	reporter := NewReporter(ctx, cancel, client, &runnerv1.Task{Context: taskCtx}, cfg)
+	reporter.ResetSteps(1)
+	reporter.RunDaemon()
+	defer func() {
+		_ = reporter.Close("")
+	}()
+
+	// Fire a log entry that starts a step — this triggers stateNotify
+	require.NoError(t, reporter.Fire(&log.Entry{
+		Message: "step starting",
+		Data:    log.Fields{"stage": "Main", "stepNumber": 0, "raw_output": true},
+	}))
+
+	// stateNotify should trigger immediate UpdateTask call
+	assert.Eventually(t, func() bool {
+		return updateTaskCalls.Load() > 0
+	}, 500*time.Millisecond, 10*time.Millisecond,
+		"step transition should have triggered immediate state flush via stateNotify")
 }