compactor: skip final status update if job not found by scheduler (#15094)

#### What this PR does

When the compaction worker is sending job status updates to the
compactor scheduler, the scheduler can reply "I have no idea what job
you are talking about" (`NotFound`) if a lease expired.

Previously, the worker would then still reply with another message to
say "Please reassign this job that you don't know about" and the
scheduler would then again reply "I have no idea what job you are
talking about".

This change makes the worker skip the useless reassignment reply by
recognizing that the job context was canceled by the status update
goroutine with a sentinel error.

#### Which issue(s) this PR fixes or relates to

Fixes N/A

#### Checklist

- [x] Tests updated.
- [ ] Documentation added.
- [ ] `CHANGELOG.md` updated - the order of entries should be
`[CHANGE]`, `[FEATURE]`, `[ENHANCEMENT]`, `[BUGFIX]`. If changelog entry
is not needed, please add the `changelog-not-needed` label to the PR.
- [ ]
[`about-versioning.md`](https://github.com/grafana/mimir/blob/main/docs/sources/mimir/configure/about-versioning.md)
updated with experimental features.

<!-- CURSOR_SUMMARY -->
---

> [!NOTE]
> **Low Risk**
> Low risk: narrowly changes scheduler-mode job status reporting to
avoid sending terminal updates after a `NotFound` lease cancellation,
with targeted tests covering both job types.
> 
> **Overview**
> When the scheduler responds `NotFound` to a periodic keep-alive, the
compactor now treats the job as *canceled by the scheduler* and **skips
sending any final status/reassign update** for that job.
> 
> This introduces a sentinel cancellation cause
(`errJobCanceledByScheduler`) and adds a new test that simulates
mid-flight cancellation for both compaction and planning jobs, asserting
only the heartbeat update is attempted and no final status is sent.
> 
> <sup>Reviewed by [Cursor Bugbot](https://cursor.com/bugbot) for commit
42ad3c7. Bugbot is set up for automated
code reviews on this repo. Configure
[here](https://www.cursor.com/dashboard/bugbot).</sup>
<!-- /CURSOR_SUMMARY -->

Author: andyasp

pkg/compactor/executor.go

@@ -42,6 +42,7 @@ import (
 var (
 	errCompactionJobHasNoBlocks = errors.New("compaction job has no blocks")
 	errNoBlockMetadataProvided  = errors.New("no block metadata provided")
+	errJobCanceledByScheduler   = errors.New("job canceled by scheduler")
 )
 
 // compactionExecutor defines how compaction work is executed.
@@ -317,7 +318,7 @@ func (e *schedulerExecutor) startJobStatusUpdater(ctx context.Context, c *Multit
 				// Check if the job was canceled from the scheduler side (not found response)
 				if grpcutil.ErrorToStatusCode(err) == codes.NotFound {
 					level.Info(e.logger).Log("msg", "job canceled by scheduler, stopping work", "job_id", jobId, "tenant", jobTenant)
-					cancelJob(err) // Cancel the job context to stop the main work
+					cancelJob(errJobCanceledByScheduler) // Cancel the job context to stop the main work
 					return
 				}
 				level.Warn(e.logger).Log("msg", "failed to send keep-alive update", "job_id", jobId, "tenant", jobTenant, "err", err)
@@ -418,7 +419,9 @@ func (e *schedulerExecutor) leaseAndExecuteJob(ctx context.Context, c *Multitena
 		wg.Wait()
 		if err != nil {
 			level.Warn(e.logger).Log("msg", "failed to execute job", "job_id", jobID, "tenant", jobTenant, "job_type", jobType, "err", err)
-			e.sendFinalJobStatus(ctx, resp.Key, resp.Spec, status)
+			if !errors.Is(context.Cause(jobCtx), errJobCanceledByScheduler) {
+				e.sendFinalJobStatus(ctx, resp.Key, resp.Spec, status)
+			}
 			return true, err
 		}
 		e.sendFinalJobStatus(ctx, resp.Key, resp.Spec, status)
@@ -430,12 +433,14 @@ func (e *schedulerExecutor) leaseAndExecuteJob(ctx context.Context, c *Multitena
 		wg.Wait()
 		if planErr != nil {
 			level.Warn(e.logger).Log("msg", "failed to execute planning job", "job_id", jobID, "tenant", jobTenant, "job_type", jobType, "err", planErr)
-			// Planning jobs only send final status updates on failure
-			e.sendFinalJobStatus(ctx, resp.Key, resp.Spec, compactorschedulerpb.UPDATE_TYPE_REASSIGN)
+			if !errors.Is(context.Cause(jobCtx), errJobCanceledByScheduler) {
+				// Only send an update on failure if the scheduler still thinks we own the job.
+				e.sendFinalJobStatus(ctx, resp.Key, resp.Spec, compactorschedulerpb.UPDATE_TYPE_REASSIGN)
+			}
 			return true, planErr
 		}
 
-		// For planning jobs, no final status update is sent - results are communicated via PlannedJobs
+		// For planning jobs, no final status update is sent on success. Completion is communicated via PlannedJobs.
 		if err := e.sendPlannedJobs(ctx, resp.Key, resp.Spec, plannedJobs); err != nil {
 			level.Warn(e.logger).Log("msg", "failed to send planned jobs", "job_id", jobID, "tenant", jobTenant, "num_jobs", len(plannedJobs), "err", err)
 			return true, err

pkg/compactor/executor_test.go

@@ -7,6 +7,7 @@ import (
 	"context"
 	"errors"
 	"fmt"
+	"io"
 	"math/rand"
 	"os"
 	"path/filepath"
@@ -920,3 +921,94 @@ func TestBuildCompactionJobFromMetas(t *testing.T) {
 		})
 	}
 }
+
+type blockingBucket struct {
+	objstore.Bucket
+}
+
+func (b *blockingBucket) Get(ctx context.Context, _ string) (io.ReadCloser, error) {
+	// Blocks compaction jobs
+	<-ctx.Done()
+	return nil, ctx.Err()
+}
+
+func (b *blockingBucket) Iter(ctx context.Context, _ string, _ func(string) error, _ ...objstore.IterOption) error {
+	// Blocks planning jobs
+	<-ctx.Done()
+	return ctx.Err()
+}
+
+func TestSchedulerExecutor_SchedulerCancellation_SkipsFinalStatus(t *testing.T) {
+	tests := map[string]struct {
+		leaseResponse *compactorschedulerpb.LeaseJobResponse
+		setupUpdate   func(*mockCompactorSchedulerClient)
+	}{
+		"compaction": {
+			leaseResponse: &compactorschedulerpb.LeaseJobResponse{
+				Key: &compactorschedulerpb.JobKey{Id: "compaction"},
+				Spec: &compactorschedulerpb.JobSpec{
+					Tenant:  "tenant",
+					Job:     &compactorschedulerpb.CompactionJob{BlockIds: [][]byte{testBlockID1.Bytes()}},
+					JobType: compactorschedulerpb.JOB_TYPE_COMPACTION,
+				},
+			},
+			setupUpdate: func(m *mockCompactorSchedulerClient) {
+				m.UpdateJobFunc = func(_ context.Context, _ *compactorschedulerpb.UpdateCompactionJobRequest) (*compactorschedulerpb.UpdateJobResponse, error) {
+					return nil, status.Error(codes.NotFound, "not found")
+				}
+			},
+		},
+		"planning": {
+			leaseResponse: &compactorschedulerpb.LeaseJobResponse{
+				Key: &compactorschedulerpb.JobKey{Id: "planning"},
+				Spec: &compactorschedulerpb.JobSpec{
+					Tenant:  "tenant",
+					JobType: compactorschedulerpb.JOB_TYPE_PLANNING,
+				},
+			},
+			setupUpdate: func(m *mockCompactorSchedulerClient) {
+				m.UpdatePlanJobFunc = func(_ context.Context, _ *compactorschedulerpb.UpdatePlanJobRequest) (*compactorschedulerpb.UpdateJobResponse, error) {
+					return nil, status.Error(codes.NotFound, "not found")
+				}
+			},
+		},
+	}
+
+	for testName, tc := range tests {
+		t.Run(testName, func(t *testing.T) {
+			mockSchedulerClient := &mockCompactorSchedulerClient{}
+			mockSchedulerClient.LeaseJobFunc = func(_ context.Context, _ *compactorschedulerpb.LeaseJobRequest) (*compactorschedulerpb.LeaseJobResponse, error) {
+				return tc.leaseResponse, nil
+			}
+			tc.setupUpdate(mockSchedulerClient)
+
+			cfg := makeTestCompactorConfig()
+			cfg.SchedulerClientConfig.UpdateInterval = 1 * time.Millisecond // arbitrary
+
+			schedulerExec := newTestSchedulerExecutor(t, cfg, mockSchedulerClient)
+			c := prepareCompactorForExecutorTest(t, cfg, &blockingBucket{Bucket: objstore.NewInMemBucket()}, newMockConfigProvider())
+
+			synctest.Test(t, func(t *testing.T) {
+				errCh := make(chan error, 1)
+				go func() {
+					_, err := schedulerExec.leaseAndExecuteJob(context.Background(), c, "compactor-1")
+					errCh <- err
+				}()
+
+				// Wait until both goroutines get caught:
+				// 1. The job is blocked in a bucket call
+				// 2. The heartbeat is waiting on its ticker
+				synctest.Wait()
+
+				// Advance synctest time so a heartbeat is sent.
+				// The scheduler will return NotFound in response, which cancels the job context,
+				// which then frees the goroutine from the blocking bucket.
+				time.Sleep(cfg.SchedulerClientConfig.UpdateInterval + time.Millisecond)
+				synctest.Wait()
+
+				require.Error(t, <-errCh)
+				require.Equal(t, 1, mockSchedulerClient.GetUpdateJobCallCount()) // no final status sent
+			})
+		})
+	}
+}