s3 lifecycle: report work-based job progress during a pass (#12644)

* s3 lifecycle: report work-based job progress during a pass

The daily pass sent progress only while starting, so a healthy scan that
ran for many hours could reach the admin's 24-hour no-progress guard.
Worker heartbeats prove the worker is alive, not that this job moves.

While dailyrun.Run is in progress the handler now sends a progress update
every 5 minutes, and only when the pass's counters (entries visited,
replay events, RPC outcomes including errors) moved since the last
delivered update. An idle tick sends nothing, so a stalled pass still
expires. A failed send is retried on the next tick. The reporter is
stopped and joined before the completion is sent, so no progress can
follow the terminal result.

Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>

* admin: test that progress, not heartbeats, keeps an old job active

Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>

* s3 lifecycle: label progress phase and bucket as the last started

Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>

---------

Co-authored-by: Claude Opus 5.5 <noreply@anthropic.com>

weed/admin/plugin/plugin_monitor_test.go

@@ -793,6 +793,64 @@ func TestExpireStaleJobsCancelsTheJobOnItsWorker(t *testing.T) {
 	expectCancelFor(t, session, "req-stale")
 }
 
+// A job older than the stale limit stays active while its worker reports
+// progress for it; the worker's heartbeat alone does not keep it alive.
+func TestExpireStaleJobsKeepsAJobThatReportsProgress(t *testing.T) {
+	t.Parallel()
+
+	pluginSvc, err := New(Options{})
+	if err != nil {
+		t.Fatalf("New: %v", err)
+	}
+	defer pluginSvc.Shutdown()
+
+	const workerID = "worker-1"
+	session := &streamSession{workerID: workerID, outgoing: make(chan *plugin_pb.AdminToWorkerMessage, 8), done: make(chan struct{})}
+	pluginSvc.putSession(session)
+	pluginSvc.handleWorkerMessage(session, &plugin_pb.WorkerToAdminMessage{
+		Body: &plugin_pb.WorkerToAdminMessage_Hello{Hello: &plugin_pb.WorkerHello{WorkerId: workerID}},
+	})
+	pluginSvc.trackExecutionStart("req-working", workerID, &plugin_pb.JobSpec{JobId: "job-working", JobType: "s3_lifecycle"}, 1)
+	pluginSvc.trackExecutionStart("req-stalled", workerID, &plugin_pb.JobSpec{JobId: "job-stalled", JobType: "s3_lifecycle"}, 1)
+
+	startedAt := time.Now().UTC().Add(-defaultStaleActiveJobTimeout - time.Hour)
+	pluginSvc.jobsMu.Lock()
+	for _, job := range pluginSvc.jobs {
+		job.CreatedAt = timeToPtr(startedAt)
+		job.UpdatedAt = timeToPtr(startedAt)
+	}
+	pluginSvc.jobsMu.Unlock()
+
+	pluginSvc.handleWorkerMessage(session, &plugin_pb.WorkerToAdminMessage{
+		Body: &plugin_pb.WorkerToAdminMessage_Heartbeat{Heartbeat: &plugin_pb.WorkerHeartbeat{
+			WorkerId:            workerID,
+			ExecutionSlotsTotal: 4,
+			ExecutionSlotsUsed:  2,
+			RunningWork: []*plugin_pb.RunningWork{
+				{WorkId: "job-working", Kind: plugin_pb.WorkKind_WORK_KIND_EXECUTION, JobType: "s3_lifecycle"},
+				{WorkId: "job-stalled", Kind: plugin_pb.WorkKind_WORK_KIND_EXECUTION, JobType: "s3_lifecycle"},
+			},
+		}},
+	})
+	pluginSvc.handleWorkerMessage(session, &plugin_pb.WorkerToAdminMessage{
+		Body: &plugin_pb.WorkerToAdminMessage_JobProgressUpdate{JobProgressUpdate: &plugin_pb.JobProgressUpdate{
+			JobId: "job-working", JobType: "s3_lifecycle", RequestId: "req-working",
+			State: plugin_pb.JobState_JOB_STATE_RUNNING, Stage: "walk", Message: "entries=10",
+		}},
+	})
+
+	if n := pluginSvc.expireStaleJobs(time.Now().UTC()); n != 1 {
+		t.Fatalf("expired %d jobs, want only the stalled one", n)
+	}
+	if got, _ := pluginSvc.GetTrackedJob("job-working"); got == nil || got.State != "job_state_running" {
+		t.Fatalf("expected the job reporting progress to stay running, got %+v", got)
+	}
+	if got, _ := pluginSvc.GetTrackedJob("job-stalled"); got == nil || got.State != StateFailed {
+		t.Fatalf("expected the heartbeat-only job to be failed, got %+v", got)
+	}
+	expectCancelFor(t, session, "req-stalled")
+}
+
 func expectNoCancel(t *testing.T, session *streamSession) {
 	t.Helper()
 	select {

weed/worker/tasks/s3_lifecycle/handler.go

@@ -347,6 +347,7 @@ func (h *Handler) executeDailyReplay(ctx context.Context, request *plugin_pb.Exe
 			len(shards), cfg.Workers, limiterDesc, len(buckets), len(parseErrors)),
 	})
 
+	stopProgress := startProgressReporter(request, sender, runStats.Summary, progressInterval)
 	report.err = dailyrun.Run(ctx, dailyrun.Config{
 		Shards:          shards,
 		BucketsPath:     bucketsPath,
@@ -363,6 +364,7 @@ func (h *Handler) executeDailyReplay(ctx context.Context, request *plugin_pb.Exe
 		ClientName:      "worker-s3-lifecycle-daily",
 		Stats:           runStats,
 	})
+	stopProgress()
 	report.canceled = ctx.Err() != nil
 	report.elapsed = time.Since(startedAt)
 	report.summary = runStats.Summary()

weed/worker/tasks/s3_lifecycle/progress.go

@@ -0,0 +1,71 @@
+package s3_lifecycle
+
+import (
+	"fmt"
+	"time"
+
+	"github.com/seaweedfs/seaweedfs/weed/glog"
+	"github.com/seaweedfs/seaweedfs/weed/pb/plugin_pb"
+	pluginworker "github.com/seaweedfs/seaweedfs/weed/plugin/worker"
+	"github.com/seaweedfs/seaweedfs/weed/s3api/s3lifecycle/dailyrun"
+)
+
+// A long pass must keep refreshing the admin's stale-job clock, but only
+// with work it actually did: an idle tick sends nothing.
+const progressInterval = 5 * time.Minute
+
+func runWork(s dailyrun.RunSummary) int64 {
+	work := s.EntriesVisited + s.EventsScanned
+	for _, n := range s.Outcomes {
+		work += n
+	}
+	return work
+}
+
+func progressMessage(s dailyrun.RunSummary) string {
+	return fmt.Sprintf("s3 lifecycle daily replay running: last_phase=%s last_bucket=%s entries=%d objects=%d events=%d rpc_done=%d rpc_errors=%d",
+		s.Phase, s.Bucket, s.EntriesVisited, s.ObjectsVisited, s.EventsScanned, s.Outcomes["DONE"], s.Outcomes["RPC_ERROR"])
+}
+
+// startProgressReporter returns stop, which returns only once no further
+// progress can be sent, so the caller can send the completion after it.
+func startProgressReporter(request *plugin_pb.ExecuteJobRequest, sender pluginworker.ExecutionSender, summarize func() dailyrun.RunSummary, interval time.Duration) (stop func()) {
+	done := make(chan struct{})
+	finished := make(chan struct{})
+	go func() {
+		defer close(finished)
+		ticker := time.NewTicker(interval)
+		defer ticker.Stop()
+		var reported int64
+		for {
+			select {
+			case <-done:
+				return
+			case <-ticker.C:
+			}
+			summary := summarize()
+			work := runWork(summary)
+			if work == reported {
+				continue
+			}
+			select {
+			case <-done:
+				return
+			default:
+			}
+			if err := sender.SendProgress(&plugin_pb.JobProgressUpdate{
+				JobId: request.Job.JobId, JobType: jobType,
+				State: plugin_pb.JobState_JOB_STATE_RUNNING, Stage: "running",
+				Message: progressMessage(summary),
+			}); err != nil {
+				glog.V(1).Infof("s3 lifecycle job %s: send progress: %v", request.Job.JobId, err)
+				continue
+			}
+			reported = work
+		}
+	}()
+	return func() {
+		close(done)
+		<-finished
+	}
+}

weed/worker/tasks/s3_lifecycle/progress_test.go

@@ -0,0 +1,131 @@
+package s3_lifecycle
+
+import (
+	"errors"
+	"sync"
+	"sync/atomic"
+	"testing"
+	"time"
+
+	"github.com/seaweedfs/seaweedfs/weed/pb/plugin_pb"
+	"github.com/seaweedfs/seaweedfs/weed/s3api/s3lifecycle/dailyrun"
+	"github.com/stretchr/testify/require"
+)
+
+type progressSender struct {
+	mu       sync.Mutex
+	progress []*plugin_pb.JobProgressUpdate
+	failures int
+	attempts int
+}
+
+func (s *progressSender) SendProgress(p *plugin_pb.JobProgressUpdate) error {
+	s.mu.Lock()
+	defer s.mu.Unlock()
+	s.attempts++
+	if s.failures > 0 {
+		s.failures--
+		return errors.New("stream closed")
+	}
+	s.progress = append(s.progress, p)
+	return nil
+}
+
+func (s *progressSender) SendCompleted(*plugin_pb.JobCompleted) error { return nil }
+
+func (s *progressSender) sent() []*plugin_pb.JobProgressUpdate {
+	s.mu.Lock()
+	defer s.mu.Unlock()
+	return append([]*plugin_pb.JobProgressUpdate(nil), s.progress...)
+}
+
+func (s *progressSender) tries() int {
+	s.mu.Lock()
+	defer s.mu.Unlock()
+	return s.attempts
+}
+
+type fakeRun struct {
+	entries, events, done, rpcErrors atomic.Int64
+}
+
+func (f *fakeRun) summary() dailyrun.RunSummary {
+	return dailyrun.RunSummary{
+		Phase:          dailyrun.PhaseWalk,
+		Bucket:         "b1",
+		EntriesVisited: f.entries.Load(),
+		EventsScanned:  f.events.Load(),
+		Outcomes:       map[string]int64{"DONE": f.done.Load(), "RPC_ERROR": f.rpcErrors.Load()},
+	}
+}
+
+func progressRequest() *plugin_pb.ExecuteJobRequest {
+	return &plugin_pb.ExecuteJobRequest{Job: &plugin_pb.JobSpec{JobId: "job-1"}}
+}
+
+const testInterval = 5 * time.Millisecond
+
+func TestProgressReporter_IdleTicksSendNothing(t *testing.T) {
+	sender := &progressSender{}
+	run := &fakeRun{}
+	stop := startProgressReporter(progressRequest(), sender, run.summary, testInterval)
+	time.Sleep(20 * testInterval)
+	stop()
+	require.Empty(t, sender.sent())
+}
+
+func TestProgressReporter_ReportsEachKindOfWork(t *testing.T) {
+	sender := &progressSender{}
+	run := &fakeRun{}
+	stop := startProgressReporter(progressRequest(), sender, run.summary, testInterval)
+	defer stop()
+
+	bumps := []*atomic.Int64{&run.entries, &run.events, &run.done, &run.rpcErrors}
+	for i, counter := range bumps {
+		counter.Add(1)
+		require.Eventually(t, func() bool { return len(sender.sent()) == i+1 }, time.Second, testInterval)
+	}
+	time.Sleep(10 * testInterval)
+	updates := sender.sent()
+	require.Len(t, updates, len(bumps), "unchanged work must not refresh the job")
+
+	last := updates[len(updates)-1]
+	require.Equal(t, "job-1", last.JobId)
+	require.Equal(t, jobType, last.JobType)
+	require.Equal(t, plugin_pb.JobState_JOB_STATE_RUNNING, last.State)
+	require.Equal(t, "running", last.Stage)
+	require.Contains(t, last.Message, "last_phase=walk last_bucket=b1")
+	require.Contains(t, last.Message, "entries=1")
+	require.Contains(t, last.Message, "events=1")
+	require.Contains(t, last.Message, "rpc_done=1")
+	require.Contains(t, last.Message, "rpc_errors=1")
+}
+
+func TestProgressReporter_RetriesAFailedSendWithoutNewWork(t *testing.T) {
+	sender := &progressSender{failures: 2}
+	run := &fakeRun{}
+	stop := startProgressReporter(progressRequest(), sender, run.summary, testInterval)
+	defer stop()
+
+	run.entries.Add(3)
+	require.Eventually(t, func() bool { return len(sender.sent()) == 1 }, time.Second, testInterval)
+	require.Equal(t, 3, sender.tries())
+	time.Sleep(10 * testInterval)
+	require.Equal(t, 3, sender.tries(), "a delivered report must not be resent")
+	require.Contains(t, sender.sent()[0].Message, "entries=3")
+}
+
+func TestProgressReporter_NoProgressAfterStop(t *testing.T) {
+	sender := &progressSender{}
+	run := &fakeRun{}
+	stop := startProgressReporter(progressRequest(), sender, run.summary, testInterval)
+	run.done.Add(1)
+	require.Eventually(t, func() bool { return len(sender.sent()) == 1 }, time.Second, testInterval)
+
+	stop()
+	tries := sender.tries()
+	run.done.Add(1)
+	run.entries.Add(1)
+	time.Sleep(10 * testInterval)
+	require.Equal(t, tries, sender.tries())
+}