diff --git a/flow/ActorCollection.actor.cpp b/flow/ActorCollection.actor.cpp
deleted file mode 100644
index 8e19bd7f557..00000000000
--- a/flow/ActorCollection.actor.cpp
+++ /dev/null
@@ -1,275 +0,0 @@
-/*
- * ActorCollection.actor.cpp
- *
- * This source file is part of the FoundationDB open source project
- *
- * Copyright 2013-2026 Apple Inc. and the FoundationDB project authors
- *
- * Licensed under the Apache License, Version 2.0 (the "License");
- * you may not use this file except in compliance with the License.
- * You may obtain a copy of the License at
- *
- *     http://www.apache.org/licenses/LICENSE-2.0
- *
- * Unless required by applicable law or agreed to in writing, software
- * distributed under the License is distributed on an "AS IS" BASIS,
- * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
- * See the License for the specific language governing permissions and
- * limitations under the License.
- */
-
-#include "flow/ActorCollection.h"
-#include "flow/IndexedSet.h"
-#include "flow/UnitTest.h"
-#include <boost/intrusive/list.hpp>
-#include "flow/actorcompiler.h" // This must be the last #include.
-
-class Runner final : public boost::intrusive::list_base_hook<>,
-                     public Callback<Void>,
-                     public FastAllocated<Runner>,
-                     NonCopyable {
-public:
-	Runner(PromiseStream<Runner*> complete, PromiseStream<Error> errors)
-	  : complete(std::move(complete)), errors(std::move(errors)) {}
-
-	~Runner() { detach(); }
-
-	void start(Future<Void> task) {
-		if (!task.isReady()) {
-			registered = true;
-			task.addCallbackAndClear(this);
-			return;
-		}
-		if (task.isError()) {
-			error(task.getError());
-		} else {
-			fire(Void());
-		}
-	}
-
-	void fire(Void const&) override {
-#ifdef ENABLE_SAMPLING
-		LineageReference callbackLineage = lineage;
-		LineageScope scope(&callbackLineage);
-#endif
-		auto output = complete;
-		detach();
-		output.send(this);
-	}
-
-	void error(Error e) override {
-#ifdef ENABLE_SAMPLING
-		LineageReference callbackLineage = lineage;
-		LineageScope scope(&callbackLineage);
-#endif
-		auto output = errors;
-		detach();
-		if (e.code() != error_code_actor_cancelled) {
-			output.send(e);
-		}
-	}
-
-private:
-	void detach() {
-		if (registered) {
-			registered = false;
-			Callback<Void>::remove();
-		}
-	}
-
-	PromiseStream<Runner*> complete;
-	PromiseStream<Error> errors;
-	bool registered = false;
-#ifdef ENABLE_SAMPLING
-	LineageReference lineage = *currentLineage;
-#endif
-};
-
-// An intrusive list of Runners, which are FastAllocated.
-using RunnerList = boost::intrusive::list<Runner, boost::intrusive::constant_time_size<false>>;
-
-// The runners list in the ActorCollection must be destroyed when the actor is destructed rather
-// than before returning or throwing
-struct RunnerListDestroyer : NonCopyable {
-	explicit RunnerListDestroyer(RunnerList* list) : list(list) {}
-
-	~RunnerListDestroyer() {
-		list->clear_and_dispose([](Runner* r) { delete r; });
-	}
-
-	RunnerList* list;
-};
-
-ACTOR Future<Void> actorCollection(FutureStream<Future<Void>> addActor,
-                                   int* pCount,
-                                   double* lastChangeTime,
-                                   double* idleTime,
-                                   double* allTime,
-                                   bool returnWhenEmptied) {
-	state RunnerList runners;
-	state RunnerListDestroyer runnersDestroyer(&runners);
-	state PromiseStream<Runner*> complete;
-	state PromiseStream<Error> errors;
-	state int count = 0;
-	if (!pCount)
-		pCount = &count;
-
-	loop choose {
-		when(Future<Void> f = waitNext(addActor)) {
-			auto runner = new Runner(complete, errors);
-			runners.insert(runners.end(), *runner);
-			runner->start(std::move(f));
-
-			++*pCount;
-			if (*pCount == 1 && lastChangeTime && idleTime && allTime) {
-				double currentTime = now();
-				*idleTime += currentTime - *lastChangeTime;
-				*allTime += currentTime - *lastChangeTime;
-				*lastChangeTime = currentTime;
-			}
-		}
-		when(Runner* runner = waitNext(complete.getFuture())) {
-			if (!--*pCount) {
-				if (lastChangeTime && idleTime && allTime) {
-					double currentTime = now();
-					*allTime += currentTime - *lastChangeTime;
-					*lastChangeTime = currentTime;
-				}
-				if (returnWhenEmptied)
-					return Void();
-			}
-			// If we didn't return then the entire list wasn't destroyed so erase/destroy runner
-			runners.erase_and_dispose(runners.iterator_to(*runner), [](Runner* r) { delete r; });
-		}
-		when(Error e = waitNext(errors.getFuture())) {
-			throw e;
-		}
-	}
-}
-
-template <class T, class U>
-struct Traceable<std::pair<T, U>> {
-	static constexpr bool value = Traceable<T>::value && Traceable<U>::value;
-	static std::string toString(const std::pair<T, U>& p) {
-		auto tStr = Traceable<T>::toString(p.first);
-		auto uStr = Traceable<U>::toString(p.second);
-		std::string result(tStr.size() + uStr.size() + 3, 'x');
-		std::copy(tStr.begin(), tStr.end(), result.begin());
-		auto iter = result.begin() + tStr.size();
-		*(iter++) = ' ';
-		*(iter++) = '-';
-		*(iter++) = ' ';
-		std::copy(uStr.begin(), uStr.end(), iter);
-		return result;
-	}
-};
-
-void forceLinkActorCollectionTests() {}
-
-// The above implementation relies on the behavior that fulfilling a promise
-// that another when clause in the same choose block is waiting on is not fired synchronously.
-TEST_CASE("/flow/actorCollection/chooseWhen") {
-	state Promise<Void> promise;
-	choose {
-		when(wait(delay(0))) {
-			promise.send(Void());
-		}
-		when(wait(promise.getFuture())) {
-			// Should be cancelled, since another when clause in this choose block has executed
-			ASSERT(false);
-		}
-	}
-	return Void();
-}
-
-ACTOR Future<Void> failIfNotCancelled() {
-	wait(delay(0));
-	ASSERT(false);
-	return Void();
-}
-
-// test contract that actors are cancelled when the actor collection is cleared
-TEST_CASE("/flow/actorCollection/testCancel") {
-	state ActorCollection actorCollection(false);
-	int actors = deterministicRandom()->randomInt(1, 1000);
-	for (int i = 0; i < actors; i++) {
-		actorCollection.add(failIfNotCancelled());
-	}
-	actorCollection.clear(false);
-	wait(delay(0));
-	return Void();
-}
-
-Future<Void> failedActor() {
-	return operation_failed();
-}
-
-TEST_CASE("/flow/actorCollection/testReady") {
-	state ActorCollection actorCollection(true);
-	actorCollection.add(Void());
-	wait(actorCollection.getResult());
-	return Void();
-}
-
-TEST_CASE("/flow/actorCollection/testReadyWhilePending") {
-	state ActorCollection actorCollection(true);
-	state Promise<Void> pending;
-	actorCollection.add(pending.getFuture());
-	actorCollection.add(Void());
-	wait(delay(0));
-	ASSERT(!actorCollection.getResult().isReady());
-	pending.send(Void());
-	wait(actorCollection.getResult());
-	return Void();
-}
-
-TEST_CASE("/flow/actorCollection/testReadyError") {
-	state ActorCollection actorCollection(false);
-	actorCollection.add(failedActor());
-	try {
-		wait(actorCollection.getResult());
-		ASSERT(false);
-	} catch (Error& e) {
-		ASSERT_EQ(e.code(), error_code_operation_failed);
-	}
-	return Void();
-}
-
-TEST_CASE("/flow/actorCollection/testPendingErrorCancels") {
-	state ActorCollection actorCollection(false);
-	state Promise<Void> pending;
-	actorCollection.add(failIfNotCancelled());
-	actorCollection.add(pending.getFuture());
-	pending.sendError(operation_failed());
-	try {
-		wait(actorCollection.getResult());
-		ASSERT(false);
-	} catch (Error& e) {
-		ASSERT_EQ(e.code(), error_code_operation_failed);
-	}
-	wait(delay(0));
-	return Void();
-}
-
-// test contract that even if the actor collection has stopped and new actors are added to the promise stream, they are
-// all cancelled when resetting actor
-TEST_CASE("/flow/actorCollection/testCancelPromiseStream") {
-	state ActorCollection actorCollection(false);
-	int actors = deterministicRandom()->randomInt(1, 500);
-	for (int i = 0; i < actors; i++) {
-		actorCollection.add(failIfNotCancelled());
-	}
-	// this actor should cause the actorCollection actor to exit, meaning the new futures just build up in the promise
-	// stream
-	actorCollection.add(failedActor());
-	for (int i = 0; i < actors; i++) {
-		actorCollection.add(failIfNotCancelled());
-	}
-	// Instead of doing actorCollection.clear(false) we reinitialize to also clear the promise stream. Otherwise on
-	// resetting the actor collection actor, the new actors will be pulled from the promise stream into the new instance
-	// Note that this test fails on the assert in failIfNotCancelled() when this is replaced with
-	// actorCollection.clear(false).
-	actorCollection = ActorCollection(false);
-	wait(delay(0));
-	return Void();
-}
diff --git a/flow/ActorCollection.cpp b/flow/ActorCollection.cpp
new file mode 100644
index 00000000000..001c8dde3db
--- /dev/null
+++ b/flow/ActorCollection.cpp
@@ -0,0 +1,744 @@
+/*
+ * ActorCollection.cpp
+ *
+ * This source file is part of the FoundationDB open source project
+ *
+ * Copyright 2013-2026 Apple Inc. and the FoundationDB project authors
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+#include "flow/ActorCollection.h"
+#include "flow/CoroUtils.h"
+#include "flow/UnitTest.h"
+#include <boost/intrusive/list.hpp>
+#include <string>
+#include <vector>
+
+#ifdef ENABLE_SAMPLING
+static LineageReference actorCollectionLineage(LineageReference const& parent) {
+	LineageReference lineage = parent;
+	lineage.setActorName("actorCollection");
+	return lineage;
+}
+#endif
+
+class ActorCollectionRuntime;
+
+// Owns a child callback and intrusive-list membership; completion may synchronously destroy it.
+class Runner final : public boost::intrusive::list_base_hook<>,
+                     public Callback<Void>,
+                     public FastAllocated<Runner>,
+                     NonCopyable {
+public:
+	explicit Runner(ActorCollectionRuntime* owner) : owner(owner) {}
+
+	~Runner() { detach(); }
+
+	void start(Future<Void> task);
+	void fire(Void const&) override;
+	void error(Error e) override;
+
+private:
+	void detach() {
+		if (registered) {
+			registered = false;
+			Callback<Void>::remove();
+		}
+	}
+
+	ActorCollectionRuntime* owner;
+	Runner* nextCompleted = nullptr;
+	bool registered = false;
+#ifdef ENABLE_SAMPLING
+	LineageReference lineage = actorCollectionLineage(*currentLineage);
+#endif
+
+	friend class ActorCollectionRuntime;
+};
+
+// An intrusive list of Runners, which are FastAllocated.
+using RunnerList = boost::intrusive::list<Runner, boost::intrusive::constant_time_size<false>>;
+
+// Coordinates queued additions, child completion, and reentrant-safe collection teardown.
+class ActorCollectionRuntime final : NonCopyable {
+	// Keeps the add callback registered; terminal delivery may synchronously destroy it.
+	class AddActorCallback final : public SingleCallback<Future<Void>> {
+	public:
+		explicit AddActorCallback(ActorCollectionRuntime* owner) : owner(owner) {}
+
+		void fire(Future<Void> const& actor) override;
+		void fire(Future<Void>&& actor) override;
+		void error(Error e) override;
+
+	private:
+		ActorCollectionRuntime* owner;
+	};
+
+public:
+	ActorCollectionRuntime(FutureStream<Future<Void>> addActor,
+	                       int* pCount,
+	                       double* lastChangeTime,
+	                       double* idleTime,
+	                       double* allTime,
+	                       bool returnWhenEmptied)
+	  : addActor(std::move(addActor)), pCount(pCount), lastChangeTime(lastChangeTime), idleTime(idleTime),
+	    allTime(allTime), returnWhenEmptied(returnWhenEmptied), addActorCallback(this) {
+		if (!this->pCount) {
+			this->pCount = &count;
+		}
+	}
+
+	~ActorCollectionRuntime() {
+		finished = true;
+		if (addCallbackRegistered) {
+			addActorCallback.remove();
+			addCallbackRegistered = false;
+		}
+		runners.clear_and_dispose([](Runner* runner) { delete runner; });
+	}
+
+	Future<Void> getResult() { return done.getFuture(); }
+	void start() {
+#ifdef ENABLE_SAMPLING
+		LineageReference callbackLineage = actorCollectionLineage(lineage);
+		LineageScope scope(&callbackLineage);
+#endif
+		drain();
+	}
+
+private:
+	void onAdded(Future<Void> actor) {
+		if (finished) {
+			return;
+		}
+		if (draining) {
+			pendingAdds.emplace_back(std::move(actor));
+			return;
+		}
+
+		draining = true;
+		handleAdded(std::move(actor));
+		draining = false;
+		drain();
+	}
+
+	void onAddError(Error e) {
+		if (!finished) {
+			requestError(e);
+			drain();
+		}
+	}
+
+	void onCompleted(Runner* runner) {
+		if (finished) {
+			return;
+		}
+		if (completedTail) {
+			completedTail->nextCompleted = runner;
+		} else {
+			completedHead = runner;
+		}
+		completedTail = runner;
+		drain();
+	}
+
+	void onError(Error e) {
+		if (finished) {
+			return;
+		}
+		if (!failure.present()) {
+			failure = e;
+		}
+		drain();
+	}
+
+	void incrementCount() {
+		++*pCount;
+		if (*pCount == 1 && lastChangeTime && idleTime && allTime) {
+			double currentTime = now();
+			*idleTime += currentTime - *lastChangeTime;
+			*allTime += currentTime - *lastChangeTime;
+			*lastChangeTime = currentTime;
+		}
+	}
+
+	void decrementCount() {
+		if (!--*pCount && lastChangeTime && idleTime && allTime) {
+			double currentTime = now();
+			*allTime += currentTime - *lastChangeTime;
+			*lastChangeTime = currentTime;
+		}
+	}
+
+	void addRunner(Future<Void> actor) {
+		auto runner = runners.insert(runners.end(), *new Runner(this));
+		// Synchronous completion stays queued until this runner has been counted.
+		runner->start(std::move(actor));
+		incrementCount();
+	}
+
+	void handleAdded(Future<Void> actor) {
+		ASSERT(draining);
+		// Completing inline must not outrun an earlier queued addition.
+		if (!runners.empty() || !actor.isReady() || addActor.isReady() || !pendingAdds.empty()) {
+			addRunner(std::move(actor));
+			return;
+		}
+
+		if (actor.isError()) {
+			Error e = actor.getError();
+			if (e.code() == error_code_actor_cancelled) {
+				// Cancelled children do not report completion and remain counted until teardown.
+				addRunner(std::move(actor));
+				return;
+			}
+			incrementCount();
+			requestError(e);
+			return;
+		}
+
+		incrementCount();
+		decrementCount();
+		if (!*pCount && returnWhenEmptied) {
+			terminalRequested = true;
+		}
+	}
+
+	void handleCompleted(Runner* runner) {
+		decrementCount();
+		if (!*pCount && returnWhenEmptied) {
+			terminalRequested = true;
+			return;
+		}
+		runners.erase_and_dispose(runners.iterator_to(*runner), [](Runner* runner) { delete runner; });
+	}
+
+	Runner* popCompleted() {
+		Runner* runner = completedHead;
+		completedHead = runner->nextCompleted;
+		if (!completedHead) {
+			completedTail = nullptr;
+		}
+		runner->nextCompleted = nullptr;
+		return runner;
+	}
+
+	void armAddCallback() {
+		if (!addCallbackRegistered) {
+			addCallbackRegistered = true;
+			auto stream = addActor;
+			stream.addCallbackAndClear(&addActorCallback);
+		}
+	}
+
+	void requestError(Error e) {
+		if (!terminalError.present()) {
+			terminalError = e;
+		}
+		terminalRequested = true;
+	}
+
+	void deliverTerminal() {
+		// Sending can synchronously destroy this runtime, so retain the promise before notifying waiters.
+		Promise<Void> terminal = done;
+		Optional<Error> error = terminalError;
+		finished = true;
+		if (addCallbackRegistered) {
+			addActorCallback.remove();
+			addCallbackRegistered = false;
+		}
+		if (error.present()) {
+			terminal.sendError(error.get());
+		} else {
+			terminal.send(Void());
+		}
+	}
+
+	void drain() {
+		if (draining || finished) {
+			return;
+		}
+		draining = true;
+
+		while (!terminalRequested) {
+			if (!pendingAdds.empty()) {
+				Future<Void> actor = std::move(pendingAdds[nextPendingAdd++]);
+				if (nextPendingAdd == pendingAdds.size()) {
+					pendingAdds.clear();
+					nextPendingAdd = 0;
+				}
+				handleAdded(std::move(actor));
+				continue;
+			}
+			if (addActor.isReady()) {
+				if (addActor.isError()) {
+					requestError(addActor.getError());
+				} else {
+					handleAdded(addActor.pop());
+				}
+				continue;
+			}
+			if (completedHead) {
+				handleCompleted(popCompleted());
+				continue;
+			}
+			if (failure.present()) {
+				requestError(failure.get());
+				continue;
+			}
+			armAddCallback();
+			draining = false;
+			return;
+		}
+
+		deliverTerminal();
+	}
+
+	FutureStream<Future<Void>> addActor;
+	int* pCount;
+	double* lastChangeTime;
+	double* idleTime;
+	double* allTime;
+	bool returnWhenEmptied;
+	int count = 0;
+	RunnerList runners;
+	Promise<Void> done;
+	AddActorCallback addActorCallback;
+	Runner* completedHead = nullptr;
+	Runner* completedTail = nullptr;
+	std::vector<Future<Void>> pendingAdds;
+	size_t nextPendingAdd = 0;
+	Optional<Error> failure;
+	Optional<Error> terminalError;
+	bool addCallbackRegistered = false;
+	bool draining = false;
+	bool terminalRequested = false;
+	bool finished = false;
+#ifdef ENABLE_SAMPLING
+	LineageReference lineage = actorCollectionLineage(*currentLineage);
+#endif
+
+	friend class Runner;
+};
+
+void Runner::start(Future<Void> task) {
+	if (!task.isReady()) {
+		registered = true;
+		task.addCallbackAndClear(this);
+		return;
+	}
+	if (task.isError()) {
+		error(task.getError());
+	} else {
+		fire(Void());
+	}
+}
+
+void Runner::fire(Void const&) {
+#ifdef ENABLE_SAMPLING
+	LineageReference callbackLineage = actorCollectionLineage(lineage);
+	LineageScope scope(&callbackLineage);
+#endif
+	ActorCollectionRuntime* runtime = owner;
+	detach();
+	runtime->onCompleted(this);
+}
+
+void Runner::error(Error e) {
+#ifdef ENABLE_SAMPLING
+	LineageReference callbackLineage = actorCollectionLineage(lineage);
+	LineageScope scope(&callbackLineage);
+#endif
+	ActorCollectionRuntime* runtime = owner;
+	detach();
+	if (e.code() != error_code_actor_cancelled) {
+		runtime->onError(e);
+	}
+}
+
+void ActorCollectionRuntime::AddActorCallback::fire(Future<Void> const& actor) {
+#ifdef ENABLE_SAMPLING
+	LineageReference callbackLineage = actorCollectionLineage(owner->lineage);
+	LineageScope scope(&callbackLineage);
+#endif
+	owner->onAdded(actor);
+}
+
+void ActorCollectionRuntime::AddActorCallback::fire(Future<Void>&& actor) {
+#ifdef ENABLE_SAMPLING
+	LineageReference callbackLineage = actorCollectionLineage(owner->lineage);
+	LineageScope scope(&callbackLineage);
+#endif
+	owner->onAdded(std::move(actor));
+}
+
+void ActorCollectionRuntime::AddActorCallback::error(Error e) {
+#ifdef ENABLE_SAMPLING
+	LineageReference callbackLineage = actorCollectionLineage(owner->lineage);
+	LineageScope scope(&callbackLineage);
+#endif
+	owner->onAddError(e);
+}
+
+// Collections that never return on empty normally end by cancellation without unwinding.
+static Future<Void> actorCollectionImpl(FutureStream<Future<Void>> addActor,
+                                        int* pCount,
+                                        double* lastChangeTime,
+                                        double* idleTime,
+                                        double* allTime,
+                                        NoThrowOnCancel = {}) {
+	ActorCollectionRuntime runtime(std::move(addActor), pCount, lastChangeTime, idleTime, allTime, false);
+	Future<Void> result = runtime.getResult();
+	runtime.start();
+	co_await result;
+}
+
+static Future<Void> actorCollectionUntilEmpty(FutureStream<Future<Void>> addActor,
+                                              int* pCount,
+                                              double* lastChangeTime,
+                                              double* idleTime,
+                                              double* allTime) {
+	ActorCollectionRuntime runtime(std::move(addActor), pCount, lastChangeTime, idleTime, allTime, true);
+	Future<Void> result = runtime.getResult();
+	runtime.start();
+	co_await result;
+}
+
+Future<Void> actorCollection(FutureStream<Future<Void>> const& addActor,
+                             int* const& pCount,
+                             double* const& lastChangeTime,
+                             double* const& idleTime,
+                             double* const& allTime,
+                             bool const& returnWhenEmptied) {
+	if (returnWhenEmptied) {
+		return actorCollectionUntilEmpty(addActor, pCount, lastChangeTime, idleTime, allTime);
+	}
+	return actorCollectionImpl(addActor, pCount, lastChangeTime, idleTime, allTime);
+}
+
+void forceLinkActorCollectionTests() {}
+
+TEST_CASE("/flow/actorCollection/chooseWhen") {
+	Promise<Void> promise;
+	co_await Choose()
+	    .When(delay(0), [&promise](Void const&) { promise.send(Void()); })
+	    .When(promise.getFuture(), [](Void const&) { ASSERT(false); })
+	    .run();
+	ASSERT(promise.isSet());
+}
+
+Future<Void> failIfNotCancelled() {
+	co_await delay(0);
+	ASSERT(false);
+}
+
+static Future<Void> recordActorCancellation(Future<Void> pending, std::vector<int>* cancelled, int index) {
+	try {
+		co_await pending;
+	} catch (Error& e) {
+		if (e.code() == error_code_actor_cancelled) {
+			cancelled->push_back(index);
+		}
+		throw;
+	}
+}
+
+static Future<Void> signalSiblingOnActorCancellation(Future<Void> pending, Promise<Void> sibling, bool sendError) {
+	try {
+		co_await pending;
+	} catch (Error& e) {
+		if (e.code() == error_code_actor_cancelled) {
+			if (sendError) {
+				sibling.sendError(operation_failed());
+			} else {
+				sibling.send(Void());
+			}
+		}
+		throw;
+	}
+}
+
+static Future<Void> recancelCollectionOnActorCancellation(Future<Void> pending,
+                                                          ActorCollection* collection,
+                                                          bool returnWhenEmptied,
+                                                          bool clearCollection,
+                                                          int* cancellationCount) {
+	try {
+		co_await pending;
+	} catch (Error& e) {
+		if (e.code() == error_code_actor_cancelled) {
+			++*cancellationCount;
+			if (clearCollection) {
+				collection->clear(returnWhenEmptied);
+			} else {
+				collection->getResult().cancel();
+			}
+		}
+		throw;
+	}
+}
+
+static Future<Void> reclearNoErrorsCollectionOnActorCancellation(Future<Void> pending,
+                                                                 ActorCollectionNoErrors* collection,
+                                                                 int* cancellationCount) {
+	try {
+		co_await pending;
+	} catch (Error& e) {
+		if (e.code() == error_code_actor_cancelled) {
+			++*cancellationCount;
+			collection->clear();
+		}
+		throw;
+	}
+}
+
+// test contract that actors are cancelled when the actor collection is cleared
+TEST_CASE("/flow/actorCollection/testCancel") {
+	ActorCollection actorCollection(false);
+	int actors = deterministicRandom()->randomInt(1, 1000);
+	for (int i = 0; i < actors; i++) {
+		actorCollection.add(failIfNotCancelled());
+	}
+	actorCollection.clear(false);
+	co_await delay(0);
+}
+
+TEST_CASE("/flow/actorCollection/testCancelOrder") {
+	constexpr int actorCount = 4;
+	for (bool returnWhenEmptied : { false, true }) {
+		ActorCollection collection(returnWhenEmptied);
+		std::vector<Promise<Void>> pending(actorCount);
+		std::vector<int> cancelled;
+		for (int index = 0; index < actorCount; ++index) {
+			collection.add(recordActorCancellation(pending[index].getFuture(), &cancelled, index));
+		}
+		collection.clear(returnWhenEmptied);
+		ASSERT_EQ(cancelled.size(), pending.size());
+		for (int index = 0; index < actorCount; ++index) {
+			ASSERT_EQ(cancelled[index], index);
+		}
+	}
+	return Void();
+}
+
+TEST_CASE("/flow/actorCollection/testCancelReentrantSiblingCompletion") {
+	for (bool returnWhenEmptied : { false, true }) {
+		for (bool sendError : { false, true }) {
+			ActorCollection collection(returnWhenEmptied);
+			Promise<Void> pending;
+			Promise<Void> sibling;
+			collection.add(signalSiblingOnActorCancellation(pending.getFuture(), sibling, sendError));
+			collection.add(sibling.getFuture());
+			collection.clear(returnWhenEmptied);
+			ASSERT(sibling.isSet());
+			ASSERT_EQ(sibling.isError(), sendError);
+		}
+	}
+	return Void();
+}
+
+TEST_CASE("/flow/actorCollection/testCancelReentrantCollection") {
+	for (bool returnWhenEmptied : { false, true }) {
+		for (bool clearCollection : { false, true }) {
+			for (bool nestedReturnWhenEmptied : { false, true }) {
+				ActorCollection collection(returnWhenEmptied);
+				Promise<Void> pending;
+				int cancellationCount = 0;
+				collection.add(recancelCollectionOnActorCancellation(
+				    pending.getFuture(), &collection, nestedReturnWhenEmptied, clearCollection, &cancellationCount));
+				collection.clear(returnWhenEmptied);
+				ASSERT_EQ(cancellationCount, 1);
+				Future<Void> replacement = collection.getResult();
+				ASSERT(!replacement.isReady());
+				collection.add(Void());
+				ASSERT_EQ(replacement.isReady(), clearCollection ? nestedReturnWhenEmptied : returnWhenEmptied);
+			}
+		}
+	}
+	return Void();
+}
+
+TEST_CASE("/flow/actorCollection/testCancelReentrantNoErrorsCollection") {
+	ActorCollectionNoErrors collection;
+	Promise<Void> pending;
+	int cancellationCount = 0;
+	collection.add(reclearNoErrorsCollectionOnActorCancellation(pending.getFuture(), &collection, &cancellationCount));
+	ASSERT_EQ(collection.size(), 1);
+
+	collection.clear();
+	ASSERT_EQ(cancellationCount, 1);
+	ASSERT_EQ(collection.size(), 0);
+
+	Promise<Void> replacement;
+	collection.add(replacement.getFuture());
+	ASSERT_EQ(collection.size(), 1);
+	replacement.send(Void());
+	ASSERT_EQ(collection.size(), 0);
+	return Void();
+}
+
+#ifdef ENABLE_SAMPLING
+// Captures the active actor lineage when a collection result is delivered synchronously.
+class ActorCollectionLineageObserver final : public Callback<Void>, NonCopyable {
+public:
+	void fire(Void const&) override { observe(); }
+
+	void error(Error e) override {
+		observedError = e;
+		observe();
+	}
+
+	void assertObserved(bool expectError) const {
+		ASSERT_EQ(actorName, std::string("actorCollection"));
+		ASSERT_EQ(observedError.present(), expectError);
+	}
+
+private:
+	void observe() {
+		actorName = currentLineage->actorName();
+		Callback<Void>::remove();
+	}
+
+	std::string actorName;
+	Optional<Error> observedError;
+};
+
+TEST_CASE("/flow/actorCollection/testSamplingLineage") {
+	for (bool readyActor : { false, true }) {
+		PromiseStream<Future<Void>> addActor;
+		ActorCollectionRuntime collection(addActor.getFuture(), nullptr, nullptr, nullptr, nullptr, true);
+		collection.start();
+
+		ActorCollectionLineageObserver observer;
+		Future<Void> result = collection.getResult();
+		result.addCallbackAndClear(&observer);
+
+		if (readyActor) {
+			addActor.send(Future<Void>(Void()));
+		} else {
+			Promise<Void> pending;
+			addActor.send(pending.getFuture());
+			pending.send(Void());
+		}
+		observer.assertObserved(false);
+	}
+
+	PromiseStream<Future<Void>> addActor;
+	ActorCollectionRuntime collection(addActor.getFuture(), nullptr, nullptr, nullptr, nullptr, false);
+	collection.start();
+	ActorCollectionLineageObserver observer;
+	Future<Void> result = collection.getResult();
+	result.addCallbackAndClear(&observer);
+	addActor.sendError(operation_failed());
+	observer.assertObserved(true);
+	return Void();
+}
+#endif
+
+Future<Void> failedActor() {
+	return operation_failed();
+}
+
+TEST_CASE("/flow/actorCollection/testReady") {
+	ActorCollection actorCollection(true);
+	actorCollection.add(Void());
+	co_await actorCollection.getResult();
+}
+
+TEST_CASE("/flow/actorCollection/testReadyWhilePending") {
+	ActorCollection actorCollection(true);
+	Promise<Void> pending;
+	actorCollection.add(pending.getFuture());
+	actorCollection.add(Void());
+	co_await delay(0);
+	ASSERT(!actorCollection.getResult().isReady());
+	pending.send(Void());
+	co_await actorCollection.getResult();
+}
+
+TEST_CASE("/flow/actorCollection/testQueuedReadyWhilePending") {
+	PromiseStream<Future<Void>> addActor;
+	Promise<Void> pending;
+	int count = 0;
+	addActor.send(Void());
+	addActor.send(pending.getFuture());
+
+	Future<Void> collection = actorCollection(addActor.getFuture(), &count, nullptr, nullptr, nullptr, true);
+	ASSERT_EQ(count, 1);
+	ASSERT(!collection.isReady());
+
+	pending.send(Void());
+	co_await collection;
+	ASSERT_EQ(count, 0);
+}
+
+TEST_CASE("/flow/actorCollection/testReadyError") {
+	ActorCollection actorCollection(false);
+	actorCollection.add(failedActor());
+	try {
+		co_await actorCollection.getResult();
+		ASSERT(false);
+	} catch (Error& e) {
+		ASSERT_EQ(e.code(), error_code_operation_failed);
+	}
+}
+
+TEST_CASE("/flow/actorCollection/testAddStreamError") {
+	PromiseStream<Future<Void>> addActor;
+	Future<Void> collection = actorCollection(addActor.getFuture());
+	addActor.sendError(operation_failed());
+	try {
+		co_await collection;
+		ASSERT(false);
+	} catch (Error& e) {
+		ASSERT_EQ(e.code(), error_code_operation_failed);
+	}
+}
+
+TEST_CASE("/flow/actorCollection/testPendingErrorCancels") {
+	ActorCollection actorCollection(false);
+	Promise<Void> pending;
+	actorCollection.add(failIfNotCancelled());
+	actorCollection.add(pending.getFuture());
+	pending.sendError(operation_failed());
+	try {
+		co_await actorCollection.getResult();
+		ASSERT(false);
+	} catch (Error& e) {
+		ASSERT_EQ(e.code(), error_code_operation_failed);
+	}
+	co_await delay(0);
+}
+
+// test contract that even if the actor collection has stopped and new actors are added to the promise stream, they are
+// all cancelled when resetting actor
+TEST_CASE("/flow/actorCollection/testCancelPromiseStream") {
+	ActorCollection actorCollection(false);
+	int actors = deterministicRandom()->randomInt(1, 500);
+	for (int i = 0; i < actors; i++) {
+		actorCollection.add(failIfNotCancelled());
+	}
+	// this actor should cause the actorCollection actor to exit, meaning the new futures just build up in the promise
+	// stream
+	actorCollection.add(failedActor());
+	for (int i = 0; i < actors; i++) {
+		actorCollection.add(failIfNotCancelled());
+	}
+	// Instead of doing actorCollection.clear(false) we reinitialize to also clear the promise stream. Otherwise on
+	// resetting the actor collection actor, the new actors will be pulled from the promise stream into the new instance
+	// Note that this test fails on the assert in failIfNotCancelled() when this is replaced with
+	// actorCollection.clear(false).
+	actorCollection = ActorCollection(false);
+	co_await delay(0);
+}
diff --git a/flow/include/flow/ActorCollection.h b/flow/include/flow/ActorCollection.h
index fae1261351c..6efb7452cb3 100644
--- a/flow/include/flow/ActorCollection.h
+++ b/flow/include/flow/ActorCollection.h
@@ -54,7 +54,11 @@ struct ActorCollectionNoErrors : NonCopyable {
 public:
 	ActorCollectionNoErrors() { init(); }
 	void clear() {
-		m_ac = Future<Void>();
+		Future<Void> previous = m_ac;
+		previous.cancel();
+		if (m_ac != previous) {
+			return;
+		}
 		init();
 	}
 	void add(Future<Void> actor) { m_add.send(actor); }
@@ -73,8 +77,13 @@ class ActorCollection : NonCopyable {
 
 	void add(Future<Void> a) { m_add.send(a); }
 	Future<Void> getResult() const { return m_out; }
+	// A reentrant clear retains the innermost replacement and its returnWhenEmptied setting.
 	void clear(bool returnWhenEmptied) {
-		m_out.cancel();
+		Future<Void> previous = m_out;
+		previous.cancel();
+		if (m_out != previous) {
+			return;
+		}
 		m_out = actorCollection(m_add.getFuture(), nullptr, nullptr, nullptr, nullptr, returnWhenEmptied);
 	}
 };