diff --git a/internal/pkg/config/config.example.yaml b/internal/pkg/config/config.example.yaml index 8d1df857..1b394b02 100644 --- a/internal/pkg/config/config.example.yaml +++ b/internal/pkg/config/config.example.yaml @@ -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. diff --git a/internal/pkg/report/reporter.go b/internal/pkg/report/reporter.go index dd44881e..e4733d72 100644 --- a/internal/pkg/report/reporter.go +++ b/internal/pkg/report/reporter.go @@ -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 diff --git a/internal/pkg/report/reporter_test.go b/internal/pkg/report/reporter_test.go index 0462a4e1..43aec80c 100644 --- a/internal/pkg/report/reporter_test.go +++ b/internal/pkg/report/reporter_test.go @@ -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() +}