runtime: close native fleet idle and successor wake races

internal/build/coro_native_fleet_e2e_test.go

@@ -321,10 +321,16 @@ import _ "unsafe"
 var Failed uint32
 var MainThread uintptr
 var RetiredThread uintptr
+var SuccessorBefore uintptr
+var SuccessorAfter uintptr
+var SuccessorDone chan uint32
 
 //go:linkname osThreadLock llgo.coroOSThreadLock
 func osThreadLock()
 
+//go:linkname timerSleep llgo.coroTimerSleep
+func timerSleep(delay int64)
+
 //llgo:coro noblock
 //go:linkname threadID C.__llgo_coro_native_fleet_e2e_thread_id_v1
 func threadID() uintptr
@@ -341,6 +347,9 @@ func Setup() {
 	Failed = 0
 	MainThread = threadID()
 	RetiredThread = 0
+	SuccessorBefore = 0
+	SuccessorAfter = 0
+	SuccessorDone = make(chan uint32)
 }
 
 func retireLockedPeer() {
@@ -358,12 +367,33 @@ func retireLockedPeer() {
 	// before publishing the G as reclaimable.
 }
 
+func wakeRetiredPeerSuccessor() {
+	thread := threadID()
+	if thread == MainThread {
+		go wakeRetiredPeerSuccessor()
+		return
+	}
+	SuccessorBefore = thread
+	// With no other peer work, this parks the clean successor's route and
+	// forces its next resume through a fresh logical owner epoch.
+	timerSleep(5 * 1000 * 1000)
+	SuccessorAfter = threadID()
+	SuccessorDone <- 1
+}
+
 func main() {
 	go retireLockedPeer()
 	for threadExitCount() == 0 {
 	}
 	if RetiredThread == 0 || RetiredThread == MainThread {
 		Failed = 42
+		return
+	}
+	go wakeRetiredPeerSuccessor()
+	<-SuccessorDone
+	if SuccessorBefore == 0 || SuccessorBefore == MainThread ||
+		SuccessorBefore == RetiredThread || SuccessorAfter != SuccessorBefore {
+		Failed = 44
 	}
 }
 

runtime/internal/coro/executor_driver.go

@@ -665,18 +665,28 @@ func (probe ExecutorWorkerCompletionProbe) Ready() bool {
 
 // PrepareExecutorWorkerCompletionProbe observes the one condition under which
 // a native target may profitably defer ArmIdle: an exact submitted worker
-// operation is still incomplete. It never turns that advisory observation into
-// a correctness obligation. A target which does not observe Ready within its
-// bounded policy must use the ordinary retained wait transaction unchanged.
+// operation is still incomplete. A direct-channel producer may also win after
+// the owner reached stable idle but before this advisory probe. Report that
+// ordinary race as ready work with an invalid probe so the target re-enters
+// the unified reducer; a valid ready probe continues to mean worker
+// completion. Every owner-only cursor, queue header, and lifecycle invariant
+// remains fail-closed. A target which does not observe Ready within its bounded
+// policy must use the ordinary retained wait transaction unchanged.
 func PrepareExecutorWorkerCompletionProbe(
 	driver *ExecutorDriver,
 ) (probe ExecutorWorkerCompletionProbe, awaiting, ready, ok bool) {
 	if !validExecutorDriver(driver) || driver.state != executorDriverActive ||
 		driver.run.issued != ActionInvalid || driver.poll.phase != executorPollIdle ||
 		!emptyOwnerLocalCompletion(&driver.local) ||
-		!executorDirectChannelInboxIdle(driver) || !idleExecutorScheduler(driver.p) {
+		!idleExecutorScheduler(driver.p) {
+		return ExecutorWorkerCompletionProbe{}, false, false, false
+	}
+	if driver.directChannelTail == nil || preemptLoadPointer(&driver.directChannelHead) == nil {
 		return ExecutorWorkerCompletionProbe{}, false, false, false
 	}
+	if !executorDirectChannelInboxIdle(driver) {
+		return ExecutorWorkerCompletionProbe{}, false, true, true
+	}
 	if driver.sources.worker == nil {
 		return ExecutorWorkerCompletionProbe{}, false, false, true
 	}

runtime/internal/coro/executor_driver_test.go

@@ -291,6 +291,53 @@ func TestPrepareExecutorStandbyDefersDirectChannelIngress(t *testing.T) {
 	closeTestExecutorDriver(t, driver)
 }
 
+func TestWorkerCompletionProbeDefersDirectChannelIngress(t *testing.T) {
+	p := new(P)
+	driver, _, _ := bindTestExecutorDriver(t, p)
+	// A fleet peer may still own cold fairness/source state here; only an
+	// issued physical action or active source transaction would make the
+	// completion window unstable.
+	driver.run.sourceMore = true
+	completion := &DirectChannelCompletion{
+		owner: driver,
+		route: driver.route,
+		state: uint32(directChannelCompletionMatched),
+	}
+	if !PublishExecutorDirectChannelCompletion(driver, completion) {
+		t.Fatal("publish direct-channel completion before worker probe")
+	}
+	probe, awaiting, ready, ok := PrepareExecutorWorkerCompletionProbe(driver)
+	if !ok || awaiting || !ready || probe.Valid() {
+		t.Fatalf(
+			"worker probe over direct-channel ingress = (%+v, %t, %t, %t), want (invalid, false, true, true)",
+			probe, awaiting, ready, ok,
+		)
+	}
+	if got, ok := takeExecutorDirectChannelCompletion(driver); !ok || got != completion {
+		t.Fatalf("take probe-deferred completion = (%p, %t), want (%p, true)", got, ok, completion)
+	}
+	if !EnterExecutorRunCompatibility(driver) {
+		t.Fatal("settle cold cursor after probe-deferred completion")
+	}
+	closeTestExecutorDriver(t, driver)
+}
+
+func TestWorkerCompletionProbeRejectsInvalidDirectChannelHeader(t *testing.T) {
+	p := new(P)
+	driver, _, _ := bindTestExecutorDriver(t, p)
+	tail := driver.directChannelTail
+	driver.directChannelTail = nil
+	probe, awaiting, ready, ok := PrepareExecutorWorkerCompletionProbe(driver)
+	if ok || awaiting || ready || probe.Valid() {
+		t.Fatalf(
+			"worker probe with invalid direct-channel header = (%+v, %t, %t, %t), want zero invalid result",
+			probe, awaiting, ready, ok,
+		)
+	}
+	driver.directChannelTail = tail
+	closeTestExecutorDriver(t, driver)
+}
+
 func TestCommitExecutorStandbyDefersDirectChannelIngress(t *testing.T) {
 	p := new(P)
 	driver, _, _, _ := bindTestExecutorDriverWithTimers(t, p)

runtime/internal/runtime/coro_native_m_owner_llgo.go

@@ -422,11 +422,26 @@ func coroNativeMActiveOwnerV1(
 			if !domain.adopted || owner.ownerEpoch != coroNativeProgramOwnerEpochV1 {
 				return nil, nil, 0, 0, false
 			}
-		} else if domain.adopted || domain.ownerEpoch == 0 ||
-			owner.ownerEpoch != domain.ownerEpoch {
-			return nil, nil, 0, 0, false
+			epoch = owner.ownerEpoch
+		} else {
+			if domain.adopted || domain.ownerEpoch == 0 {
+				return nil, nil, 0, 0, false
+			}
+			if owner.baton.Valid() {
+				// A replacement successor inherits one released execution-domain
+				// baton and therefore remains bound to that exact logical epoch.
+				if owner.ownerEpoch != domain.ownerEpoch {
+					return nil, nil, 0, 0, false
+				}
+				epoch = owner.ownerEpoch
+			} else {
+				// An ordinary clean successor is the permanent physical M for this
+				// route. Like the initial peer above it spans arbitrarily many
+				// idle/wake epochs, so the live domain—not its creation record—is
+				// authoritative for the current logical owner epoch.
+				epoch = domain.ownerEpoch
+			}
 		}
-		epoch = owner.ownerEpoch
 	default:
 		return nil, nil, 0, 0, false
 	}