admin/plugin: fail a dispatch its restarted worker no longer lists (#12402)

* admin/plugin: fail a dispatch its restarted worker no longer lists

A job dispatched to a worker that then restarts dies with the old
process. The worker re-registers under the same id on a new stream and
its heartbeats stop listing the job, but nothing ended the pending
execution: the dispatcher kept waiting and held the executor slot until
the attempt's execution timeout.

Each pending execution now records the stream it went out on. An
itemized heartbeat from a newer stream of that worker that leaves the
job out twice in a row fails it through the expire path, so the
dispatch returns, the slot is released and the retry machinery takes
over. Jobs on the heartbeat's own stream, jobs the new stream still
lists, and heartbeats that count slots without itemizing them are left
alone.

Claude-Session: https://claude.ai/code/session_011UpvafcfUuC2mLxNAXLQJi

* admin/plugin: let a partly itemized listing clear a dispatch's misses

A heartbeat that counted more slots than it itemized returned before the
listed jobs had their misses reset, so an omission, a partial heartbeat
listing the job, and another omission failed a job the worker was still
running.

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

* admin/plugin: look up the worker's stream before tracking the attempt

A dispatch to a worker with no stream returned after the job was already
recorded as assigned, and that record kept new proposals with the same
dedupe key out until the stale-job sweep expired it.

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

* admin/plugin: fail the attempt whose request could not be sent

The dispatch now sends on the stream it looked up before tracking the
attempt, so a worker that reconnects in between makes the send fail.
That returned with the job still tracked as assigned, and with the
worker back nothing closed the record for the 24h stale timeout while
its dedupe key kept new proposals out. The attempt is now completed as
failed, after it is unregistered so the completion does not wait on a
channel nobody reads.

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

* admin/plugin: leave a send failure to the expiry that already claimed it

The send-error path and an orphan expiry could both complete the same
attempt and record two failed runs for it. The send-error path now
completes the attempt only if no expiry has marked it failed, decided
under the same lock that unregisters it.

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

* admin/plugin: record a send failure whose entry Shutdown already removed

Shutdown closes shutdownCh before it unregisters pending dispatches, so a
blocked send returns while its map entry is gone. Read the expiry claim
from the dispatch's own state instead of the map lookup, so the attempt
is still failed and does not stay assigned.

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

* admin/plugin: match a padded job id against its trimmed listing

Heartbeat work ids are trimmed before the orphan check, so compare the
dispatch's job id trimmed too; otherwise a job id with surrounding
spaces never matches its own listing and the running job is failed.

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

* admin/plugin: skip orphan misses once a completion was handed over

The buffer length only shows a completion the dispatcher has not taken
yet; after it takes one, the attempt stays registered until the dispatch
returns. Mark the hand-off under the pending lock and check that flag
instead, so heartbeats in that gap cannot fail a finished attempt.

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

---------

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

weed/admin/plugin/orphaned_execution.go

@@ -0,0 +1,90 @@
+package plugin
+
+import (
+	"fmt"
+	"strings"
+	"time"
+
+	"github.com/seaweedfs/seaweedfs/weed/glog"
+	"github.com/seaweedfs/seaweedfs/weed/pb/plugin_pb"
+)
+
+// orphanedExecutionMisses is how many itemized heartbeats in a row from a
+// newer stream of the worker must leave a dispatched job out before the
+// admin fails it. The second rides out a heartbeat the worker built while
+// it was still picking the job back up.
+const orphanedExecutionMisses = 2
+
+// pendingExecutionState is an execution request waiting for its JobCompleted.
+type pendingExecutionState struct {
+	completed chan *plugin_pb.JobCompleted
+	jobID     string
+	jobType   string
+	// session is the stream the request went out on.
+	session *streamSession
+	misses  int
+	failed  bool
+	// completionSent is set once a completion is handed to the dispatcher,
+	// which may already have taken it from the buffer.
+	completionSent bool
+}
+
+// failExecutionsMissingFromHeartbeat fails the executions sent on an earlier
+// stream of the heartbeat's worker that the worker no longer lists. A worker
+// that restarted comes back under the same id without the old process's
+// jobs, and nothing else ends such a dispatch before its execution timeout,
+// so its slot stays reserved all that time. Jobs sent on the heartbeat's own
+// stream are left alone, since a worker counts a job before it lists it, and
+// so is everything while the heartbeat counts slots it does not itemize,
+// since a missing id then proves nothing.
+func (r *Plugin) failExecutionsMissingFromHeartbeat(session *streamSession, heartbeat *plugin_pb.WorkerHeartbeat) {
+	if session == nil || heartbeat == nil {
+		return
+	}
+	itemized := 0
+	listed := make(map[string]struct{}, len(heartbeat.GetRunningWork()))
+	for _, work := range heartbeat.GetRunningWork() {
+		if work.GetKind() == plugin_pb.WorkKind_WORK_KIND_EXECUTION {
+			listed[strings.TrimSpace(work.GetWorkId())] = struct{}{}
+			itemized++
+		}
+	}
+	itemizesAll := int(heartbeat.GetExecutionSlotsUsed()) <= itemized
+
+	var orphaned []activeJobSnapshot
+	r.pendingExecutionMu.Lock()
+	for requestID, pending := range r.pendingExecution {
+		if pending.failed || pending.session == nil || pending.session == session || pending.session.workerID != session.workerID {
+			continue
+		}
+		if _, ok := listed[strings.TrimSpace(pending.jobID)]; ok {
+			pending.misses = 0
+			continue
+		}
+		if !itemizesAll {
+			continue
+		}
+		if pending.completionSent {
+			continue
+		}
+		pending.misses++
+		if pending.misses < orphanedExecutionMisses {
+			continue
+		}
+		pending.failed = true
+		orphaned = append(orphaned, activeJobSnapshot{
+			jobID:     pending.jobID,
+			jobType:   pending.jobType,
+			workerID:  session.workerID,
+			requestID: requestID,
+		})
+	}
+	r.pendingExecutionMu.Unlock()
+
+	for _, snap := range orphaned {
+		reason := fmt.Sprintf("worker %s re-registered without job %s", snap.workerID, snap.jobID)
+		glog.Warningf("Plugin failing %s request %s: %s", snap.jobType, snap.requestID, reason)
+		// off the stream's receive loop: the cancel may wait on a full buffer
+		go r.expireActiveJob(snap, reason, time.Now().UTC())
+	}
+}

weed/admin/plugin/orphaned_execution_test.go

@@ -0,0 +1,389 @@
+package plugin
+
+import (
+	"context"
+	"strings"
+	"testing"
+	"time"
+
+	"github.com/seaweedfs/seaweedfs/weed/pb/plugin_pb"
+)
+
+const orphanTestJobType = "ec_vacuum"
+
+type orphanTestResult struct {
+	completed *plugin_pb.JobCompleted
+	err       error
+}
+
+func connectExecutor(p *Plugin, workerID string) *streamSession {
+	p.registry.UpsertFromHello(&plugin_pb.WorkerHello{
+		WorkerId:         workerID,
+		WorkerInstanceId: workerID + "-instance-" + time.Now().Format("150405.000000"),
+		Capabilities:     []*plugin_pb.JobTypeCapability{{JobType: orphanTestJobType, CanExecute: true, MaxExecutionConcurrency: 2}},
+	})
+	session := &streamSession{workerID: workerID, outgoing: make(chan *plugin_pb.AdminToWorkerMessage, 8), done: make(chan struct{})}
+	p.putSession(session)
+	return session
+}
+
+// dispatchToExecutor runs one execution on session's worker and returns the
+// request the worker was sent along with the dispatch's eventual result.
+func dispatchToExecutor(t *testing.T, p *Plugin, session *streamSession, jobID string) (string, <-chan orphanTestResult) {
+	t.Helper()
+	executor, ok := p.registry.Get(session.workerID)
+	if !ok {
+		t.Fatalf("worker %s is not registered", session.workerID)
+	}
+	resultCh := make(chan orphanTestResult, 1)
+	go func() {
+		ctx, cancel := context.WithTimeout(context.Background(), time.Hour)
+		defer cancel()
+		completed, err := p.executeJobWithExecutor(ctx, executor, &plugin_pb.JobSpec{JobId: jobID, JobType: orphanTestJobType}, &plugin_pb.ClusterContext{}, 1)
+		resultCh <- orphanTestResult{completed: completed, err: err}
+	}()
+	select {
+	case msg := <-session.outgoing:
+		if msg.GetExecuteJobRequest() == nil {
+			t.Fatalf("expected an execute_job_request, got %+v", msg)
+		}
+		return msg.RequestId, resultCh
+	case <-time.After(2 * time.Second):
+		t.Fatal("no execute request")
+		return "", nil
+	}
+}
+
+func sendExecutionHeartbeat(p *Plugin, session *streamSession, slotsUsed int32, jobIDs ...string) {
+	heartbeat := &plugin_pb.WorkerHeartbeat{WorkerId: session.workerID, ExecutionSlotsUsed: slotsUsed, ExecutionSlotsTotal: 2}
+	for _, jobID := range jobIDs {
+		heartbeat.RunningWork = append(heartbeat.RunningWork, &plugin_pb.RunningWork{
+			WorkId:  jobID,
+			Kind:    plugin_pb.WorkKind_WORK_KIND_EXECUTION,
+			JobType: orphanTestJobType,
+			State:   plugin_pb.JobState_JOB_STATE_RUNNING,
+		})
+	}
+	p.handleWorkerMessage(session, &plugin_pb.WorkerToAdminMessage{Body: &plugin_pb.WorkerToAdminMessage_Heartbeat{Heartbeat: heartbeat}})
+}
+
+func expectDispatchStillPending(t *testing.T, resultCh <-chan orphanTestResult) {
+	t.Helper()
+	select {
+	case got := <-resultCh:
+		t.Fatalf("dispatch ended early: completed=%+v err=%v", got.completed, got.err)
+	case <-time.After(100 * time.Millisecond):
+	}
+}
+
+// The worker process restarted after the job was dispatched: it comes back
+// under the same id on a new stream and its heartbeats no longer list the
+// job. The dispatch fails instead of holding its slot until the execution
+// timeout.
+func TestExecutionFailsWhenWorkerRestartsWithoutJob(t *testing.T) {
+	t.Parallel()
+	p, err := New(Options{})
+	if err != nil {
+		t.Fatalf("New: %v", err)
+	}
+	defer p.Shutdown()
+
+	const jobID = "job-lost-in-restart"
+	requestID, resultCh := dispatchToExecutor(t, p, connectExecutor(p, "worker-a"), jobID)
+	restarted := connectExecutor(p, "worker-a")
+
+	sendExecutionHeartbeat(p, restarted, 0)
+	expectDispatchStillPending(t, resultCh)
+	sendExecutionHeartbeat(p, restarted, 0)
+
+	select {
+	case got := <-resultCh:
+		if got.err == nil || got.completed == nil || got.completed.Success {
+			t.Fatalf("expected a failed completion, got completed=%+v err=%v", got.completed, got.err)
+		}
+		if want := "worker worker-a re-registered without job " + jobID; !strings.Contains(got.err.Error(), want) {
+			t.Fatalf("error %q does not say %q", got.err, want)
+		}
+	case <-time.After(5 * time.Second):
+		t.Fatal("dispatch was not failed after its worker restarted without the job")
+	}
+
+	tracked, ok := p.GetTrackedJob(jobID)
+	if !ok || tracked == nil {
+		t.Fatalf("job %s is not tracked", jobID)
+	}
+	if tracked.State != StateFailed || tracked.RequestID != requestID {
+		t.Fatalf("expected request %s failed, got state=%s request=%s", requestID, tracked.State, tracked.RequestID)
+	}
+}
+
+// A worker that kept the job running across a reconnect lists it from the
+// new stream; the dispatch keeps waiting and takes the completion from there.
+func TestExecutionKeptWhenWorkerReconnectsWithJob(t *testing.T) {
+	t.Parallel()
+	p, err := New(Options{})
+	if err != nil {
+		t.Fatalf("New: %v", err)
+	}
+	defer p.Shutdown()
+
+	const jobID = "job-kept-across-reconnect"
+	requestID, resultCh := dispatchToExecutor(t, p, connectExecutor(p, "worker-a"), jobID)
+	reconnected := connectExecutor(p, "worker-a")
+
+	for i := 0; i < orphanedExecutionMisses+1; i++ {
+		sendExecutionHeartbeat(p, reconnected, 1, jobID)
+	}
+	expectDispatchStillPending(t, resultCh)
+
+	p.handleWorkerMessage(reconnected, &plugin_pb.WorkerToAdminMessage{Body: &plugin_pb.WorkerToAdminMessage_JobCompleted{JobCompleted: &plugin_pb.JobCompleted{
+		RequestId: requestID,
+		JobId:     jobID,
+		JobType:   orphanTestJobType,
+		Success:   true,
+	}}})
+	select {
+	case got := <-resultCh:
+		if got.err != nil || got.completed == nil || !got.completed.Success {
+			t.Fatalf("expected success, got completed=%+v err=%v", got.completed, got.err)
+		}
+	case <-time.After(5 * time.Second):
+		t.Fatal("completion from the reconnected stream did not reach the dispatch")
+	}
+}
+
+// Work ids are compared trimmed, so a padded job id still matches its listing.
+func TestExecutionKeptWhenReconnectedWorkerListsPaddedJobID(t *testing.T) {
+	t.Parallel()
+	p, err := New(Options{})
+	if err != nil {
+		t.Fatalf("New: %v", err)
+	}
+	defer p.Shutdown()
+
+	const jobID = " job-padded "
+	_, resultCh := dispatchToExecutor(t, p, connectExecutor(p, "worker-a"), jobID)
+	reconnected := connectExecutor(p, "worker-a")
+
+	for i := 0; i < orphanedExecutionMisses+1; i++ {
+		sendExecutionHeartbeat(p, reconnected, 1, jobID)
+	}
+	expectDispatchStillPending(t, resultCh)
+}
+
+// The worker counts a job before it lists it, and heartbeats can lag; the
+// stream the job went out on is never judged by what it has not listed yet.
+func TestExecutionKeptWhileDispatchSessionHasNotListedJob(t *testing.T) {
+	t.Parallel()
+	p, err := New(Options{})
+	if err != nil {
+		t.Fatalf("New: %v", err)
+	}
+	defer p.Shutdown()
+
+	session := connectExecutor(p, "worker-a")
+	_, resultCh := dispatchToExecutor(t, p, session, "job-just-dispatched")
+
+	for i := 0; i < orphanedExecutionMisses+1; i++ {
+		sendExecutionHeartbeat(p, session, 0)
+		sendExecutionHeartbeat(p, session, 1)
+	}
+	expectDispatchStillPending(t, resultCh)
+}
+
+// A heartbeat that counts busy slots without itemizing them (the Rust worker
+// reports only the count) cannot show a job is gone.
+func TestExecutionKeptWhenHeartbeatDoesNotItemize(t *testing.T) {
+	t.Parallel()
+	p, err := New(Options{})
+	if err != nil {
+		t.Fatalf("New: %v", err)
+	}
+	defer p.Shutdown()
+
+	_, resultCh := dispatchToExecutor(t, p, connectExecutor(p, "worker-a"), "job-count-only")
+	reconnected := connectExecutor(p, "worker-a")
+
+	for i := 0; i < orphanedExecutionMisses+1; i++ {
+		sendExecutionHeartbeat(p, reconnected, 1)
+		sendExecutionHeartbeat(p, reconnected, 2, "some-other-job")
+	}
+	expectDispatchStillPending(t, resultCh)
+}
+
+// A listing between two omissions shows the worker still runs the job, even
+// when that heartbeat also counts a slot it does not itemize.
+func TestExecutionKeptWhenReplacementStreamListsJobBetweenOmissions(t *testing.T) {
+	t.Parallel()
+	for name, listing := range map[string]int32{"itemized": 1, "partly itemized": 2} {
+		t.Run(name, func(t *testing.T) {
+			p, err := New(Options{})
+			if err != nil {
+				t.Fatalf("New: %v", err)
+			}
+			defer p.Shutdown()
+
+			const jobID = "job-listed-between-omissions"
+			_, resultCh := dispatchToExecutor(t, p, connectExecutor(p, "worker-a"), jobID)
+			reconnected := connectExecutor(p, "worker-a")
+
+			sendExecutionHeartbeat(p, reconnected, 0)
+			sendExecutionHeartbeat(p, reconnected, listing, jobID)
+			sendExecutionHeartbeat(p, reconnected, 0)
+			expectDispatchStillPending(t, resultCh)
+		})
+	}
+}
+
+func TestExecutionNotTrackedWhenWorkerNotConnected(t *testing.T) {
+	t.Parallel()
+	p, err := New(Options{})
+	if err != nil {
+		t.Fatalf("New: %v", err)
+	}
+	defer p.Shutdown()
+
+	p.registry.UpsertFromHello(&plugin_pb.WorkerHello{
+		WorkerId:     "worker-a",
+		Capabilities: []*plugin_pb.JobTypeCapability{{JobType: orphanTestJobType, CanExecute: true, MaxExecutionConcurrency: 2}},
+	})
+	executor, ok := p.registry.Get("worker-a")
+	if !ok {
+		t.Fatal("worker-a is not registered")
+	}
+	const jobID = "job-without-session"
+	if _, err := p.executeJobWithExecutor(context.Background(), executor, &plugin_pb.JobSpec{JobId: jobID, JobType: orphanTestJobType}, &plugin_pb.ClusterContext{}, 1); err == nil {
+		t.Fatal("expected an error for a worker without a stream")
+	}
+	if tracked, ok := p.GetTrackedJob(jobID); ok && tracked != nil {
+		t.Fatalf("job %s tracked as %s for a dispatch that never went out", jobID, tracked.State)
+	}
+}
+
+// The captured stream closed before the request went out, as when the worker
+// reconnects mid-dispatch: no report will ever close the attempt.
+func TestExecutionFailedWhenSendToSessionFails(t *testing.T) {
+	t.Parallel()
+	p, err := New(Options{})
+	if err != nil {
+		t.Fatalf("New: %v", err)
+	}
+	defer p.Shutdown()
+
+	p.registry.UpsertFromHello(&plugin_pb.WorkerHello{
+		WorkerId:     "worker-a",
+		Capabilities: []*plugin_pb.JobTypeCapability{{JobType: orphanTestJobType, CanExecute: true, MaxExecutionConcurrency: 2}},
+	})
+	closed := &streamSession{workerID: "worker-a", outgoing: make(chan *plugin_pb.AdminToWorkerMessage), done: make(chan struct{})}
+	closed.close()
+	p.putSession(closed)
+	executor, ok := p.registry.Get("worker-a")
+	if !ok {
+		t.Fatal("worker-a is not registered")
+	}
+	const jobID = "job-send-failed"
+	if _, err := p.executeJobWithExecutor(context.Background(), executor, &plugin_pb.JobSpec{JobId: jobID, JobType: orphanTestJobType}, &plugin_pb.ClusterContext{}, 1); err == nil {
+		t.Fatal("expected an error for a closed stream")
+	}
+	tracked, ok := p.GetTrackedJob(jobID)
+	if !ok || tracked == nil || tracked.State != StateFailed {
+		t.Fatalf("expected job %s failed, got %+v", jobID, tracked)
+	}
+}
+
+// Shutdown unregisters every pending dispatch before a blocked send returns
+// its error; the attempt is still recorded as failed.
+func TestExecutionFailedWhenSendFailsAfterShutdownUnregisters(t *testing.T) {
+	t.Parallel()
+	p, err := New(Options{})
+	if err != nil {
+		t.Fatalf("New: %v", err)
+	}
+	defer p.Shutdown()
+
+	p.registry.UpsertFromHello(&plugin_pb.WorkerHello{
+		WorkerId:     "worker-a",
+		Capabilities: []*plugin_pb.JobTypeCapability{{JobType: orphanTestJobType, CanExecute: true, MaxExecutionConcurrency: 2}},
+	})
+	blocked := &streamSession{workerID: "worker-a", outgoing: make(chan *plugin_pb.AdminToWorkerMessage), done: make(chan struct{})}
+	p.putSession(blocked)
+	executor, ok := p.registry.Get("worker-a")
+	if !ok {
+		t.Fatal("worker-a is not registered")
+	}
+	const jobID = "job-send-after-shutdown"
+	errCh := make(chan error, 1)
+	go func() {
+		_, err := p.executeJobWithExecutor(context.Background(), executor, &plugin_pb.JobSpec{JobId: jobID, JobType: orphanTestJobType}, &plugin_pb.ClusterContext{}, 1)
+		errCh <- err
+	}()
+
+	deadline := time.Now().Add(2 * time.Second)
+	for unregistered := false; !unregistered; {
+		if time.Now().After(deadline) {
+			t.Fatal("dispatch never registered")
+		}
+		p.pendingExecutionMu.Lock()
+		for requestID, pending := range p.pendingExecution {
+			if pending.jobID == jobID {
+				close(pending.completed)
+				delete(p.pendingExecution, requestID)
+				unregistered = true
+			}
+		}
+		p.pendingExecutionMu.Unlock()
+		time.Sleep(time.Millisecond)
+	}
+	blocked.close()
+
+	select {
+	case err := <-errCh:
+		if err == nil {
+			t.Fatal("expected an error for a closed stream")
+		}
+	case <-time.After(2 * time.Second):
+		t.Fatal("dispatch did not return")
+	}
+	tracked, ok := p.GetTrackedJob(jobID)
+	if !ok || tracked == nil || tracked.State != StateFailed {
+		t.Fatalf("expected job %s failed, got %+v", jobID, tracked)
+	}
+}
+
+// The dispatcher can take the completion from the buffer before it
+// unregisters the attempt; heartbeats in that gap must not count misses.
+func TestExecutionNotOrphanedAfterDispatcherTookCompletion(t *testing.T) {
+	t.Parallel()
+	p, err := New(Options{})
+	if err != nil {
+		t.Fatalf("New: %v", err)
+	}
+	defer p.Shutdown()
+
+	const jobID = "job-completion-taken"
+	const requestID = "exec-completion-taken"
+	original := connectExecutor(p, "worker-a")
+	pending := &pendingExecutionState{
+		completed: make(chan *plugin_pb.JobCompleted, 1),
+		jobID:     jobID,
+		jobType:   orphanTestJobType,
+		session:   original,
+	}
+	p.pendingExecutionMu.Lock()
+	p.pendingExecution[requestID] = pending
+	p.pendingExecutionMu.Unlock()
+
+	p.safeSendJobCompleted(requestID, &plugin_pb.JobCompleted{RequestId: requestID, JobId: jobID, JobType: orphanTestJobType, Success: true})
+	<-pending.completed
+
+	reconnected := connectExecutor(p, "worker-a")
+	for i := 0; i < orphanedExecutionMisses+1; i++ {
+		sendExecutionHeartbeat(p, reconnected, 0)
+	}
+	p.pendingExecutionMu.Lock()
+	failed := pending.failed
+	p.pendingExecutionMu.Unlock()
+	if failed {
+		t.Fatal("attempt whose completion was already delivered was failed as orphaned")
+	}
+}

weed/admin/plugin/plugin.go

@@ -106,7 +106,7 @@ type Plugin struct {
 	pendingDetection   map[string]*pendingDetectionState
 
 	pendingExecutionMu sync.Mutex
-	pendingExecution   map[string]chan *plugin_pb.JobCompleted
+	pendingExecution   map[string]*pendingExecutionState
 
 	pendingPreviewMu sync.Mutex
 	pendingPreview   map[string]chan *plugin_pb.ObjectPreviewResponse
@@ -216,7 +216,7 @@ func New(options Options) (*Plugin, error) {
 		sessions:                  make(map[string]*streamSession),
 		pendingSchema:             make(map[string]chan *plugin_pb.ConfigSchemaResponse),
 		pendingDetection:          make(map[string]*pendingDetectionState),
-		pendingExecution:          make(map[string]chan *plugin_pb.JobCompleted),
+		pendingExecution:          make(map[string]*pendingExecutionState),
 		pendingPreview:            make(map[string]chan *plugin_pb.ObjectPreviewResponse),
 		observations:              NewObservationStore(),
 		nextDetectionAt:           make(map[string]time.Time),
@@ -289,8 +289,8 @@ func (r *Plugin) Shutdown() {
 	r.pendingDetectionMu.Unlock()
 
 	r.pendingExecutionMu.Lock()
-	for requestID, ch := range r.pendingExecution {
-		close(ch)
+	for requestID, pending := range r.pendingExecution {
+		close(pending.completed)
 		delete(r.pendingExecution, requestID)
 	}
 	r.pendingExecutionMu.Unlock()
@@ -815,9 +815,24 @@ func (r *Plugin) executeJobWithExecutor(
 	// the single active worker, not 1/N of it.
 	clusterContext = r.decorateClusterContextForJob(clusterContext, job.JobType, adminConfigValues, int(adminRuntime.GetMaxJobsPerDetection()))
 
+	// The request goes out on this stream and no other, so a newer stream
+	// of the worker can tell it never received it.
+	session, ok := r.getSession(executor.WorkerID)
+	if !ok {
+		return nil, fmt.Errorf("worker %s is not connected", executor.WorkerID)
+	}
+
+	r.trackExecutionStart(requestID, executor.WorkerID, job, attempt)
+
 	completedCh := make(chan *plugin_pb.JobCompleted, 1)
+	pending := &pendingExecutionState{
+		completed: completedCh,
+		jobID:     job.JobId,
+		jobType:   job.JobType,
+		session:   session,
+	}
 	r.pendingExecutionMu.Lock()
-	r.pendingExecution[requestID] = completedCh
+	r.pendingExecution[requestID] = pending
 	r.pendingExecutionMu.Unlock()
 	defer func() {
 		r.pendingExecutionMu.Lock()
@@ -825,8 +840,6 @@ func (r *Plugin) executeJobWithExecutor(
 		r.pendingExecutionMu.Unlock()
 	}()
 
-	r.trackExecutionStart(requestID, executor.WorkerID, job, attempt)
-
 	message := &plugin_pb.AdminToWorkerMessage{
 		RequestId: requestID,
 		SentAt:    timestamppb.Now(),
@@ -843,7 +856,23 @@ func (r *Plugin) executeJobWithExecutor(
 		},
 	}
 
-	if err := r.sendToWorker(executor.WorkerID, message); err != nil {
+	if err := r.sendToSession(session, message); err != nil {
+		// The worker never got the request, so no report will close the
+		// attempt unless an expiry already claimed it. Unregister first:
+		// nobody reads completedCh any more. Shutdown may have done so.
+		r.pendingExecutionMu.Lock()
+		delete(r.pendingExecution, requestID)
+		claimed := pending.failed
+		r.pendingExecutionMu.Unlock()
+		if !claimed {
+			r.handleJobCompleted(&plugin_pb.JobCompleted{
+				JobId:        job.JobId,
+				JobType:      job.JobType,
+				RequestId:    requestID,
+				ErrorMessage: err.Error(),
+				CompletedAt:  timestamppb.Now(),
+			})
+		}
 		return nil, err
 	}
 
@@ -1157,6 +1186,7 @@ func (r *Plugin) handleWorkerMessage(session *streamSession, message *plugin_pb.
 			return
 		}
 		r.resendUndeliveredCancels(workerID, body.Heartbeat)
+		r.failExecutionsMissingFromHeartbeat(session, body.Heartbeat)
 	case *plugin_pb.WorkerToAdminMessage_ConfigSchemaResponse:
 		r.handleConfigSchemaResponse(body.ConfigSchemaResponse)
 	case *plugin_pb.WorkerToAdminMessage_DetectionProposals:
@@ -1501,8 +1531,12 @@ func (r *Plugin) handleJobCompleted(completed *plugin_pb.JobCompleted) {
 }
 
 func (r *Plugin) safeSendJobCompleted(requestID string, completed *plugin_pb.JobCompleted) {
+	var ch chan *plugin_pb.JobCompleted
 	r.pendingExecutionMu.Lock()
-	ch := r.pendingExecution[requestID]
+	if pending := r.pendingExecution[requestID]; pending != nil {
+		ch = pending.completed
+		pending.completionSent = true
+	}
 	r.pendingExecutionMu.Unlock()
 	safeSendCh(ch, completed, r.shutdownCh)
 }