fix: report step log ranges with the log flush (#1189)

Gitea slices a task's single log stream into steps by the LogIndex/LogLength that UpdateTask carries. The reporter's daemon flushed log rows without those counters, which only left on the separate state ticker, so rows the server acked between two state reports belonged to no step. The web UI attributes them to no step while the job runs, and a runner that stops reporting in that window leaves them under "Complete job" for good once Gitea finalizes the task. The three log-flush paths now report the state describing the rows the server just took, so the gap is at most one RPC. Live output also stops waiting up to `state_report_interval` to become visible. An idle job adds no requests. This does not explain why the runner in that job stopped reporting for 13m40s until Gitea reaped the task as a zombie, that needs the runner host's own log. Fixes https://gitea.com/gitea/runner/issues/1184 --------- Co-authored-by: silverwind <2021+silverwind@noreply.gitea.com> Co-authored-by: silverwind <me@silverwind.io> Reviewed-on: https://gitea.com/gitea/runner/pulls/1189 Reviewed-by: silverwind <2021+silverwind@noreply.gitea.com> Co-authored-by: bircni <bircni@icloud.com>
2026-08-25 13:27:46 +00:00 · 2026-08-25 12:13:18 +00:00
parent 745a1e70e6
commit 34df4887af
3 changed files with 147 additions and 9 deletions
@@ -71,7 +71,7 @@ runner:
  # 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 is also reported on step transitions (start/stop) and after each log flush.
  #state_report_interval: 5s
  # Per-attempt deadline for flushing the final logs and task state when a job
  # finishes, on a detached context so a server cancel can't block the acknowledgement.
@@ -373,7 +373,7 @@ func (r *Reporter) runDaemonLoop() {
 	for {
 		select {
 		case <-logTicker.C:
-			r.noteReport(metrics.LabelMethodUpdateLog, r.ReportLog(false))
+			r.reportLogWithState()
 			r.stopLatencyTimer(&maxLatencyActive, maxLatencyTimer)

 		case <-stateTicker.C:
@@ -385,7 +385,7 @@ func (r *Reporter) runDaemonLoop() {
 			r.stateMu.RUnlock()

 			if n >= r.logBatchSize {
-				r.noteReport(metrics.LabelMethodUpdateLog, r.ReportLog(false))
+				r.reportLogWithState()
 				r.stopLatencyTimer(&maxLatencyActive, maxLatencyTimer)
 			} else if !maxLatencyActive && n > 0 {
 				maxLatencyTimer.Reset(r.logReportMaxLatency)
@@ -400,7 +400,7 @@ func (r *Reporter) runDaemonLoop() {

 		case <-maxLatencyTimer.C:
 			maxLatencyActive = false
-			r.noteReport(metrics.LabelMethodUpdateLog, r.ReportLog(false))
+			r.reportLogWithState()

 		case <-r.ctx.Done():
 			// Stop heartbeating on cancel so Gitea sees the runner as offline
@@ -423,6 +423,17 @@ func (r *Reporter) runDaemonLoop() {
 	}
 }

+// Gitea slices the single log stream into steps by the LogIndex/LogLength the state
+// carries, so rows acked without a state report behind them stay unattributed, and stay
+// that way for good if the runner never reports again.
+func (r *Reporter) reportLogWithState() {
+	took, err := r.reportLog(false)
+	r.noteReport(metrics.LabelMethodUpdateLog, err)
+	if took {
+		r.noteReport(metrics.LabelMethodUpdateTask, r.ReportState(false))
+	}
+}
+
 func (r *Reporter) Logf(format string, a ...any) {
 	r.stateMu.Lock()
 	defer r.stateMu.Unlock()
@@ -573,6 +584,12 @@ func (r *Reporter) rpcCtx() (context.Context, context.CancelFunc) {
 }

 func (r *Reporter) ReportLog(noMore bool) error {
+	_, err := r.reportLog(noMore)
+	return err
+}
+
+// reportLog also reports whether the server took rows it had not taken before.
+func (r *Reporter) reportLog(noMore bool) (bool, error) {
 	r.clientM.Lock()
 	defer r.clientM.Unlock()

@@ -581,7 +598,7 @@ func (r *Reporter) ReportLog(noMore bool) error {
 	r.stateMu.RUnlock()

 	if !noMore && len(rows) == 0 {
-		return nil
+		return false, nil
 	}

 	rpcCtx, rpcCancel := r.rpcCtx()
@@ -598,19 +615,20 @@ func (r *Reporter) ReportLog(noMore bool) error {
 	if err != nil {
 		metrics.ReportLogTotal.WithLabelValues(metrics.LabelResultError).Inc()
 		metrics.ClientErrors.WithLabelValues(metrics.LabelMethodUpdateLog).Inc()
-		return err
+		return false, err
 	}
 	metrics.ReportLogTotal.WithLabelValues(metrics.LabelResultSuccess).Inc()

 	ack := int(resp.Msg.AckIndex)
 	if ack < r.logOffset {
-		return errors.New("submitted logs are lost")
+		return false, errors.New("submitted logs are lost")
 	}

 	r.stateMu.Lock()
 	submitted := r.logOffset + len(rows)
 	// A server can ack beyond what it was sent; clamp to stay within the buffer.
 	ack = min(ack, submitted)
+	took := ack > r.logOffset
 	r.logRows = r.logRows[ack-r.logOffset:]
 	r.logOffset = ack
 	remaining := len(r.logRows)
@@ -621,10 +639,10 @@ func (r *Reporter) ReportLog(noMore bool) error {
 	}

 	if noMore && ack < submitted {
-		return errors.New("not all logs are submitted")
+		return took, errors.New("not all logs are submitted")
 	}

-	return nil
+	return took, nil
 }

 // ReportState only reports the job result if reportResult is true
@@ -1296,3 +1296,123 @@ func TestReporter_NoteReport(t *testing.T) {
 	require.Len(t, hook.AllEntries(), 2)
 	assert.Contains(t, hook.LastEntry().Message, "reconnected")
 }
+
+// giteaLogModel mirrors how Gitea stores a task log: UpdateLog appends rows to one
+// stream, and UpdateTask overwrites the per-step ranges the web UI slices that stream by
+// (modules/actions/task_state.go, FullSteps).
+type giteaLogModel struct {
+	mu      sync.Mutex
+	rows    int64
+	claimed int64
+}
+
+func (m *giteaLogModel) appendRows(n int) int64 {
+	m.mu.Lock()
+	defer m.mu.Unlock()
+	m.rows += int64(n)
+	return m.rows
+}
+
+func (m *giteaLogModel) applyState(state *runnerv1.TaskState) {
+	m.mu.Lock()
+	defer m.mu.Unlock()
+	for _, s := range state.GetSteps() {
+		m.claimed = max(m.claimed, s.LogIndex+s.LogLength)
+	}
+}
+
+// snapshot returns the accepted rows and how far the reported step ranges reach into
+// them. The remainder is what the web UI renders under "Complete job".
+func (m *giteaLogModel) snapshot() (rows, claimed int64) {
+	m.mu.Lock()
+	defer m.mu.Unlock()
+	return m.rows, m.claimed
+}
+
+// Regression test for https://gitea.com/gitea/runner/issues/1184.
+func TestReporter_StepRangeCoversAckedRows(t *testing.T) {
+	server := &giteaLogModel{}
+
+	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) {
+			return connect_go.NewResponse(&runnerv1.UpdateLogResponse{
+				AckIndex: server.appendRows(len(req.Msg.Rows)),
+			}), nil
+		},
+	)
+	client.On("UpdateTask", mock.Anything, mock.Anything).Return(
+		func(_ context.Context, req *connect_go.Request[runnerv1.UpdateTaskRequest]) (*connect_go.Response[runnerv1.UpdateTaskResponse], error) {
+			server.applyState(req.Msg.State)
+			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)
+
+	// Only the batch threshold may flush: the tickers are the periodic repair this test
+	// must not depend on, and Close() is the one after the damage is done.
+	cfg, _ := config.LoadDefault("")
+	cfg.Runner.LogReportInterval = time.Hour
+	cfg.Runner.LogReportMaxLatency = time.Hour
+	cfg.Runner.StateReportInterval = time.Hour
+	cfg.Runner.LogReportBatchSize = 5
+
+	reporter := NewReporter(ctx, cancel, client, &runnerv1.Task{Context: taskCtx}, cfg)
+	reporter.ResetSteps(1)
+	reporter.RunDaemon()
+	defer func() {
+		_ = reporter.Close("")
+	}()
+
+	stepData := log.Fields{"stage": "Main", "stepNumber": 0, "raw_output": true}
+
+	// The step's first row is a transition, so it flushes state as well as logs.
+	require.NoError(t, reporter.Fire(&log.Entry{Message: "step starting", Data: stepData}))
+	require.Eventually(t, func() bool {
+		rows, claimed := server.snapshot()
+		return rows == 1 && claimed == rows
+	}, time.Second, 10*time.Millisecond, "the step transition should have flushed both log and state")
+
+	for i := range cfg.Runner.LogReportBatchSize {
+		require.NoError(t, reporter.Fire(&log.Entry{
+			Message: fmt.Sprintf("step output %d", i),
+			Data:    stepData,
+		}))
+	}
+
+	assert.Eventually(t, func() bool {
+		rows, claimed := server.snapshot()
+		return rows == 1+int64(cfg.Runner.LogReportBatchSize) && claimed == rows
+	}, time.Second, 10*time.Millisecond,
+		"every log row the server accepted must be covered by a reported step range")
+}
+
+// A flush the server took no rows for describes nothing new, so it must not spend an
+// UpdateTask. UpdateTask is never registered on the client, so calling it fails the test.
+func TestReporter_FlushWithoutAckedRowsSkipsState(t *testing.T) {
+	ctx, cancel := context.WithCancel(context.Background())
+	defer cancel()
+	taskCtx, err := structpb.NewStruct(map[string]any{})
+	require.NoError(t, err)
+
+	client := mocks.NewClient(t)
+	client.On("UpdateLog", mock.Anything, mock.Anything).Once().Return(
+		func(_ context.Context, _ *connect_go.Request[runnerv1.UpdateLogRequest]) (*connect_go.Response[runnerv1.UpdateLogResponse], error) {
+			return connect_go.NewResponse(&runnerv1.UpdateLogResponse{AckIndex: 0}), nil
+		},
+	)
+
+	cfg, _ := config.LoadDefault("")
+	reporter := NewReporter(ctx, cancel, client, &runnerv1.Task{Context: taskCtx}, cfg)
+
+	// An idle job: no rows, so not even an UpdateLog.
+	reporter.reportLogWithState()
+
+	// A row the server declines to take: an UpdateLog, but still no step range to report.
+	reporter.Logf("row")
+	reporter.reportLogWithState()
+}