diff --git a/src/workerd/server/cluster/cluster.rs b/src/workerd/server/cluster/cluster.rs
index 46ee4a9c4b7..c4cf7e66e9f 100644
--- a/src/workerd/server/cluster/cluster.rs
+++ b/src/workerd/server/cluster/cluster.rs
@@ -180,8 +180,13 @@ impl Cluster {
         }
         Ok(match claim {
             Claim::Owned(owned) => {
-                let epoch = owned.epoch;
                 let key = format!("{ns}/{actor}");
+                // The node may have started draining while the claim awaited the bucket: then
+                // its objects are being let go, and one opened now would outlive that.
+                if !self.ownership.may_keep.get() {
+                    return Err(crate::ownership::draining(&key));
+                }
+                let epoch = owned.epoch;
                 self.claims.borrow_mut().insert(key, owned);
                 Route::Owned(epoch)
             }
@@ -388,11 +393,20 @@ impl Cluster {
         }
     }
 
+    /// Stops claiming before the claim reaches the bucket, but keeps the
+    /// lease: the objects this node owns stay its own, their commits still
+    /// acknowledged, while they flush before `start_draining`. One its record
+    /// names that is not open here still opens here.
+    pub fn stop_claiming(&self) {
+        self.ownership.may_claim.set(false);
+    }
+
     /// Stops claiming before the claim reaches the bucket, and parks the
     /// lease loop; then `deregister`.
     pub fn start_draining(&self) {
         self.lease.borrow_mut().draining = true;
         self.ownership.may_claim.set(false);
+        self.ownership.may_keep.set(false);
     }
 
     /// The first renewal, before the node serves.
@@ -450,12 +464,138 @@ async fn check_conditional_writes(bucket: &Bucket, node: &str) -> Result<()> {
 
 #[cfg(test)]
 mod tests {
+    use std::cell::Cell;
+    use std::cell::RefCell;
+    use std::rc::Rc;
+
+    use cluster_logic::lease;
     use futures::executor::block_on;
 
+    use super::Cluster;
+    use super::Route;
+    use super::Setup;
     use super::check_conditional_writes;
+    use super::now_ms;
     use crate::Bucket;
     use crate::bucket::memory::Memory;
 
+    /// A memory bucket that runs `hook` as it serves the `nth` read of an
+    /// owner record, from 1.
+    struct Hooked {
+        inner: Memory,
+        nth: Cell<usize>,
+        hook: RefCell<Option<Box<dyn FnOnce()>>>,
+    }
+
+    #[async_trait::async_trait(?Send)]
+    impl crate::Store for Rc<Hooked> {
+        async fn get(
+            &self,
+            key: &str,
+            to: Option<&std::path::Path>,
+        ) -> crate::Result<Option<(Vec<u8>, crate::Version)>> {
+            if key.ends_with("own.json") {
+                self.nth.set(self.nth.get().saturating_sub(1));
+                if self.nth.get() == 0
+                    && let Some(hook) = self.hook.borrow_mut().take()
+                {
+                    hook();
+                }
+            }
+            self.inner.get(key, to).await
+        }
+
+        async fn put(
+            &self,
+            key: &str,
+            body: crate::bucket::Payload,
+            guard: Option<crate::Guard>,
+            metadata: &[(&str, String)],
+        ) -> crate::Result<Option<crate::Version>> {
+            self.inner.put(key, body, guard, metadata).await
+        }
+
+        async fn list(&self, prefix: &str) -> crate::Result<Vec<crate::bucket::Entry>> {
+            self.inner.list(prefix).await
+        }
+
+        async fn delete(&self, key: &str) -> crate::Result<()> {
+            self.inner.delete(key).await
+        }
+    }
+
+    fn node(bucket: &Bucket, name: &str) -> Rc<Cluster> {
+        let setup = Setup {
+            url: String::new(),
+            options: Vec::new(),
+            node: name.to_owned(),
+            address: format!("{name}:1"),
+            ttl_ms: lease::LEASE_MS,
+            key: Vec::new(),
+        };
+        Cluster::over(bucket.clone(), setup)
+    }
+
+    #[test]
+    fn a_node_that_stopped_claiming_keeps_its_objects_and_its_lease() {
+        block_on(async {
+            let bucket = Bucket::memory("");
+            let (a, b) = (node(&bucket, "a"), node(&bucket, "b"));
+            a.join().await.unwrap();
+            b.join().await.unwrap();
+            let kept = a.acquire_or_route("ns", "kept", now_ms()).await.unwrap();
+            assert!(matches!(kept, Route::Owned(_)));
+
+            a.stop_claiming();
+            // Nothing new, nor anything it would have to take over...
+            assert!(a.acquire_or_route("ns", "new", now_ms()).await.is_err());
+            // ...but the lease loop is not parked, and the lease still renews,
+            assert!(!a.lease.borrow().draining);
+            a.renew(now_ms()).await.unwrap();
+            assert!(a.lease_holds());
+            // so a peer routes to what it owns rather than claiming it,
+            match b.acquire_or_route("ns", "kept", now_ms()).await.unwrap() {
+                Route::Remote { node, .. } => assert_eq!(node, "a"),
+                Route::Owned(_) => panic!("a peer claimed an object of a live node"),
+            }
+            // and it still takes up what it owns.
+            let again = a.acquire_or_route("ns", "kept", now_ms()).await.unwrap();
+            assert!(matches!(again, Route::Owned(_)));
+
+            // Draining, it lets its objects go, its own too.
+            a.start_draining();
+            assert!(a.lease.borrow().draining);
+            assert!(a.acquire_or_route("ns", "kept", now_ms()).await.is_err());
+        });
+    }
+
+    #[test]
+    fn a_node_that_starts_draining_mid_claim_does_not_keep_the_object() {
+        block_on(async {
+            let hooked = Rc::new(Hooked {
+                inner: Memory::default(),
+                nth: Cell::new(usize::MAX),
+                hook: RefCell::new(None),
+            });
+            let bucket = Bucket::new(Rc::clone(&hooked), "");
+            let a = node(&bucket, "a");
+            a.join().await.unwrap();
+            let kept = a.acquire_or_route("ns", "kept", now_ms()).await.unwrap();
+            assert!(matches!(kept, Route::Owned(_)));
+            // Not open here any more: taking it up again moves it to the
+            // next epoch.
+            a.close_actor("ns", "kept");
+            a.stop_claiming();
+
+            // That reads its record twice: to find it is this node's, then
+            // to advance it. The drain starts between the two.
+            let drainer = Rc::clone(&a);
+            *hooked.hook.borrow_mut() = Some(Box::new(move || drainer.start_draining()));
+            hooked.nth.set(2);
+            assert!(a.acquire_or_route("ns", "kept", now_ms()).await.is_err());
+        });
+    }
+
     #[test]
     fn a_bucket_that_ignores_preconditions_is_refused() {
         block_on(async {
diff --git a/src/workerd/server/cluster/ownership.rs b/src/workerd/server/cluster/ownership.rs
index ab0051084d9..ed3a849c3c4 100644
--- a/src/workerd/server/cluster/ownership.rs
+++ b/src/workerd/server/cluster/ownership.rs
@@ -79,6 +79,9 @@ pub struct Ownership {
     registry: Rc<Registry>,
     /// Whether this instance may claim objects: off once it drains.
     pub may_claim: Cell<bool>,
+    /// Whether this instance still takes up the objects its owner records
+    /// name, while it holds its lease: off once it lets them go.
+    pub may_keep: Cell<bool>,
     /// Where each routed object's owner was, and when that was read.
     routes: RefCell<HashMap<String, Route>>,
     /// The record each claim wrote, and the guard that it is still the
@@ -105,7 +108,7 @@ const ROUTE_REFRESH_MS: u64 = 5_000;
 /// How many times a claim retries when another instance keeps winning.
 const ATTEMPTS: usize = 8;
 
-fn draining(key: &str) -> Error {
+pub fn draining(key: &str) -> Error {
     Error::Replica(format!("{key}: this node is draining"))
 }
 
@@ -119,6 +122,7 @@ impl Ownership {
             bucket,
             registry,
             may_claim: Cell::new(true),
+            may_keep: Cell::new(true),
             routes: RefCell::new(HashMap::new()),
             claimed: RefCell::new(HashMap::new()),
         }
@@ -163,8 +167,9 @@ impl Ownership {
                 Some((body, version)) => {
                     let record: OwnerRecord = serde_json::from_slice(&body)?;
                     if record.node == me {
-                        // A draining node lets its objects go, its own too.
-                        if !self.may_claim.get() {
+                        // A node that stopped claiming still wakes what it
+                        // owns, until it lets its objects go.
+                        if !self.may_keep.get() {
                             return Err(draining(&key));
                         }
                         return Ok(Claim::Owned(record.owned()));
diff --git a/src/workerd/server/factory/worker-factory-actor.c++ b/src/workerd/server/factory/worker-factory-actor.c++
index c0a38f7ab5e..e7ebf53c2da 100644
--- a/src/workerd/server/factory/worker-factory-actor.c++
+++ b/src/workerd/server/factory/worker-factory-actor.c++
@@ -5,6 +5,7 @@
 #include "worker-factory-impl.h"
 
 #include <workerd/api/actor-state.h>
+#include <workerd/api/global-scope.h>
 #include <workerd/io/actor-cache.h>
 #include <workerd/io/io-context.h>
 #include <workerd/server/actor-id-impl.h>
@@ -458,6 +459,80 @@ class ActorHooksImpl final: public Worker::Actor::Loopback,
   }
 };
 
+// A draining cluster instance's call of an actor's `drain()` method, the actor's chance to flush
+// what it keeps in memory while the instance still owns it. A class without the method gets nothing
+// (`ran` stays false). The event completes once what the method wrote is durable: in a cluster, a
+// commit is durable once the bucket has it.
+class DrainCustomEvent final: public WorkerInterface::CustomEvent {
+ public:
+  explicit DrainCustomEvent(bool& ran): ran(ran) {}
+
+  kj::Promise<Result> run(kj::Own<IoContext_IncomingRequest> incomingRequest,
+      kj::Maybe<kj::StringPtr> entrypointName,
+      kj::Maybe<Worker::VersionInfo> versionInfo,
+      Frankenvalue props,
+      kj::TaskSet& waitUntilTasks,
+      bool isDynamicDispatch) override {
+    auto& context = incomingRequest->getContext();
+    incomingRequest->delivered("drain"_kj);
+    KJ_DEFER({ incomingRequest->drain(waitUntilTasks, kj::mv(incomingRequest)); });
+
+    EventOutcome outcome = EventOutcome::OK;
+    try {
+      co_await context.run([this, entrypointName, versionInfo = kj::mv(versionInfo),
+                               props = kj::mv(props), isDynamicDispatch](
+                               Worker::Lock& lock, IoContext& ioctx) mutable -> kj::Promise<void> {
+        KJ_IF_SOME(handler,
+            lock.getExportedHandler(entrypointName, kj::mv(versionInfo), kj::mv(props),
+                ioctx.getActor(), isDynamicDispatch)) {
+          jsg::Lock& js = lock;
+          auto self = jsg::JsObject(handler->self.getHandle(js));
+          KJ_IF_SOME(method, self.get(js, "drain"_kj).tryCast<jsg::JsFunction>()) {
+            ran = true;
+            v8::LocalVector<v8::Value> args(js.v8Isolate);
+            return ioctx.awaitJs(js, js.tryCatch([&] {
+              return js.toVoidPromise(method.call(js, self, args));
+            }, [&](jsg::Value exception) { return js.rejectedPromise<void>(kj::mv(exception)); }));
+          }
+        }
+        return kj::READY_NOW;
+      });
+      // Nothing to wait for from an actor whose method did not run.
+      if (ran) co_await context.waitForOutputLocks();
+    } catch (kj::Exception& e) {
+      incomingRequest->getMetrics().reportFailure(e);
+      context.logUncaughtExceptionAsync(
+          UncaughtExceptionSource::ASYNC_TASK, e.clone(), *incomingRequest);
+      outcome = EventOutcome::EXCEPTION;
+    }
+    co_return Result{.outcome = outcome};
+  }
+
+  kj::Promise<Result> sendRpc(capnp::HttpOverCapnpFactory&,
+      capnp::ByteStreamFactory&,
+      FrankenvalueHandler&,
+      rpc::EventDispatcher::Client) override {
+    // An instance drains only the actors it runs.
+    KJ_FAIL_REQUIRE("a drain does not cross between instances");
+  }
+
+  kj::Promise<Result> notSupported() override {
+    KJ_FAIL_REQUIRE("the target does not take a drain");
+  }
+
+  uint16_t getType() override {
+    // Defined here: the event never leaves workerd.
+    return 0x8001;
+  }
+
+  tracing::EventInfo getEventInfo() const override {
+    return tracing::CustomEventInfo();
+  }
+
+ private:
+  bool& ran;
+};
+
 // Schedules a root actor's alarms with the namespace's scheduler, at once, and in a cluster indexes
 // them after `priorTask`, as production's alarm manager orders its requests: the index takes them
 // in the order they were made.
@@ -773,6 +848,59 @@ kj::Maybe<kj::Rc<HibernationManager>> actor_hibernation_manager(const ActorHandl
       [](HibernationManager& m) { return kj::Rc<HibernationManager>(m.addRef()); });
 }
 
+kj::Promise<DrainMethod> actor_drain_method(const ActorHandle& handle) {
+  auto& actor = handle.getActor();
+  auto asyncLock = co_await actor.getWorker().takeAsyncLockWithoutRequest(nullptr);
+  // An actor still being constructed, or whose construction failed, can only tell in a request.
+  auto method = DrainMethod::Unknown;
+  actor.getWorker().runInLockScope(asyncLock, [&](Worker::Lock& lock) {
+    JSG_WITHIN_CONTEXT_SCOPE(lock, lock.getContext(), [&](jsg::Lock& js) {
+      kj::runCatchingExceptions([&]() {
+        // The constructed instance; outside a request, anything else throws.
+        KJ_IF_SOME(handler,
+            lock.getExportedHandler(kj::none, kj::none, Frankenvalue(), actor, false)) {
+          // Only a data property that is a function counts, found along the prototype chain
+          // without running the app's code: an accessor or a Proxy leaves it to the request.
+          method = js.tryCatch([&] {
+            auto context = js.v8Context();
+            v8::Local<v8::String> name = js.strIntern("drain"_kj);
+            v8::Local<v8::String> get = js.strIntern("get"_kj);
+            v8::Local<v8::String> set = js.strIntern("set"_kj);
+            v8::Local<v8::String> value = js.strIntern("value"_kj);
+            v8::Local<v8::Value> current = handler->self.getHandle(js);
+            for (auto depth = 0; depth < 64 && current->IsObject(); ++depth) {
+              auto object = current.As<v8::Object>();
+              if (object->IsProxy()) return DrainMethod::Unknown;
+              auto descriptor = jsg::check(object->GetOwnPropertyDescriptor(context, name));
+              if (descriptor->IsObject()) {
+                auto fields = descriptor.As<v8::Object>();
+                if (jsg::check(fields->HasOwnProperty(context, get)) ||
+                    jsg::check(fields->HasOwnProperty(context, set))) {
+                  return DrainMethod::Unknown;
+                }
+                return jsg::check(fields->Get(context, value))->IsFunction() ? DrainMethod::Some
+                                                                             : DrainMethod::None;
+              }
+              current = object->GetPrototype();
+            }
+            return current->IsNull() ? DrainMethod::None : DrainMethod::Unknown;
+          }, [&](jsg::Value) { return DrainMethod::Unknown; });
+        } else {
+          method = DrainMethod::None;
+        }
+      });
+    });
+  });
+  co_return method;
+}
+
+kj::Promise<bool> worker_drain(kj::Own<WorkerInterface> worker) {
+  bool ran = false;
+  auto result = co_await worker->customEvent(kj::heap<DrainCustomEvent>(ran));
+  KJ_REQUIRE(result.outcome == EventOutcome::OK, "the Durable Object's drain() failed");
+  co_return ran;
+}
+
 kj::Promise<FacetStartInfo> facet_start_resolve(kj::Own<FacetStart> start) {
   auto info = co_await (*start)();
   co_await info.ensureAllResolved();
diff --git a/src/workerd/server/factory/worker-factory.h b/src/workerd/server/factory/worker-factory.h
index 62e3706720e..ca881dadd3a 100644
--- a/src/workerd/server/factory/worker-factory.h
+++ b/src/workerd/server/factory/worker-factory.h
@@ -99,6 +99,7 @@ struct ActorStorageSpec;
 struct HeaderEdit;
 struct UdpDatagram;
 enum class AlarmElsewhere : uint8_t;
+enum class DrainMethod : uint8_t;
 
 // How the config message is read: configs can legitimately be very large and are not malicious.
 constexpr capnp::ReaderOptions CONFIG_READER_OPTIONS = {.traversalLimitInWords = kj::maxValue};
@@ -217,6 +218,8 @@ kj::Own<WorkerInterface> worker_interface_attach(
     kj::Own<WorkerInterface> inner, ::rust::Box<KeepAlive> keep);
 kj::Promise<::rust::Vec<uint64_t>> worker_bench(
     kj::Own<WorkerInterface> worker, ::rust::Slice<const uint64_t> params);
+kj::Promise<bool> worker_drain(kj::Own<WorkerInterface> worker);
+kj::Promise<DrainMethod> actor_drain_method(const ActorHandle& actor);
 kj::Own<ActorHandle> worker_new_actor(const CompiledWorker& worker,
     ::rust::Str className,
     kj::Maybe<kj::Own<Frankenvalue>> props,
diff --git a/src/workerd/server/server/actor.rs b/src/workerd/server/server/actor.rs
index 20f4d74666f..d9d7b2fdb2c 100644
--- a/src/workerd/server/server/actor.rs
+++ b/src/workerd/server/server/actor.rs
@@ -27,6 +27,7 @@ use std::rc::Weak;
 use std::time::Duration;
 
 use futures::FutureExt;
+use futures::StreamExt;
 use futures::future::Either;
 use futures::future::LocalBoxFuture;
 use futures::future::Shared;
@@ -69,6 +70,8 @@ use crate::tasks::TaskHandle;
 const IDLE_SHUTDOWN_DELAY: Duration = Duration::from_secs(10);
 /// How long a container with no clients stays in its namespace after its last access.
 const CONTAINER_EXPIRATION: Duration = Duration::from_secs(70);
+/// How many actors of a namespace run their `drain()` at once.
+const DRAIN_CONCURRENCY: usize = 32;
 /// How long a test eviction waits for the actor's requests to drain.
 const EVICT_TIMEOUT: Duration = Duration::from_secs(30);
 /// How long a test eviction waits before re-checking an actor that has no requests but is still
@@ -109,6 +112,19 @@ pub struct ActorNamespace {
     resolving: RefCell<HashMap<String, Rc<Notify>>>,
 }
 
+/// What `drain_all` did with a namespace's running actors.
+#[derive(Clone, Copy, Debug, Default, PartialEq, Eq)]
+pub struct Drained {
+    /// Their `drain()` completed, and what it wrote is durable.
+    pub drained: usize,
+    /// Their class has no `drain()`.
+    pub skipped: usize,
+    /// Their `drain()` threw, or the request it runs in failed.
+    pub failed: usize,
+    /// Cut off at the deadline.
+    pub timed_out: usize,
+}
+
 /// The first resolution of an actor: its end, however it ends, wakes the requests that waited.
 struct Resolving<'a> {
     pending: &'a RefCell<HashMap<String, Rc<Notify>>>,
@@ -373,6 +389,53 @@ impl ActorNamespace {
         }
     }
 
+    /// Whether a draining cluster instance calls this namespace's actors' `drain()`
+    /// (`DurableObjectNamespace.drain`).
+    pub fn drains(&self) -> bool {
+        matches!(self.config, ActorConfig::Durable { drain: true, .. })
+    }
+
+    /// The worker service the class belongs to.
+    pub fn service(&self) -> &str {
+        &self.service
+    }
+
+    /// Runs the `drain()` method of every running root actor whose class has one, at most
+    /// `DRAIN_CONCURRENCY` at a time, and cuts off those not done `timeout` from now. The actors
+    /// keep serving meanwhile, and stay once drained: the caller lets them go (`abort_all`).
+    pub async fn drain_all(&self, timeout: Duration) -> Drained {
+        // Collected, so that the map is not borrowed while the actors run: one that breaks
+        // leaves it.
+        let drains: Vec<_> = self
+            .actors
+            .borrow()
+            .values()
+            .filter(|container| container.is_running() && container.require_not_broken().is_ok())
+            .map(|container| {
+                let container = Rc::clone(container);
+                async move { container.drain().await }.boxed_local()
+            })
+            .collect();
+        let total = drains.len();
+        let mut results = futures::stream::iter(drains).buffer_unordered(DRAIN_CONCURRENCY);
+        let mut deadline = std::pin::pin!(self.factory.sleep(timeout));
+        let mut drained = Drained::default();
+        loop {
+            match futures::future::select(results.next(), deadline.as_mut()).await {
+                Either::Left((Some(Ok(true)), _)) => drained.drained += 1,
+                Either::Left((Some(Ok(false)), _)) => drained.skipped += 1,
+                Either::Left((Some(Err(_)), _)) => drained.failed += 1,
+                Either::Left((None, _)) => break,
+                // Dropping the drains cancels their methods' requests, and those not started.
+                Either::Right(_) => {
+                    drained.timed_out = total - drained.drained - drained.skipped - drained.failed;
+                    break;
+                }
+            }
+        }
+        drained
+    }
+
     /// Aborts every actor and deletes the namespace's storage and alarms.
     pub fn delete_all(&self, reason: Option<&crate::Error>) -> Result<()> {
         // The databases are reset while their connections are open, before the abort closes
@@ -1002,6 +1065,20 @@ impl ActorContainer {
         Ok(())
     }
 
+    /// Runs the actor's `drain()` method as a request of its own: whether its instance has one.
+    /// It completes once what the method wrote is durable. An actor without one gets no request,
+    /// so it does not wait behind the actor's input gate.
+    async fn drain(self: &Rc<Self>) -> Result<bool> {
+        let running = self.actor.borrow().clone();
+        if let Some(actor) = running
+            && ffi::actor_drain_method(&actor).await? == ffi::DrainMethod::None
+        {
+            return Ok(false);
+        }
+        let metadata = ffi::new_request_metadata(KjMaybe::None, KjMaybe::None);
+        Ok(ffi::worker_drain(self.start_request(metadata)?).await?)
+    }
+
     /// Starts a request on the actor, starting the actor first if it is not running. The request
     /// is a client of the container from here until its interface is dropped.
     pub fn start_request(
diff --git a/src/workerd/server/server/bindings-test.rs b/src/workerd/server/server/bindings-test.rs
index 18ce2ce603b..4613faad694 100644
--- a/src/workerd/server/server/bindings-test.rs
+++ b/src/workerd/server/server/bindings-test.rs
@@ -54,6 +54,7 @@ fn durable(unique_key: &str) -> ActorConfig {
         enable_sql: true,
         workflow: None,
         container: None,
+        drain: false,
     }
 }
 
diff --git a/src/workerd/server/server/bindings.rs b/src/workerd/server/server/bindings.rs
index 690731ab4f5..5638dbce552 100644
--- a/src/workerd/server/server/bindings.rs
+++ b/src/workerd/server/server/bindings.rs
@@ -118,6 +118,8 @@ pub enum ActorConfig {
         workflow: Option<Box<Workflow>>,
         /// Where the config has the class's `container` options, if it has any.
         container: Option<crate::bridge::ffi::ContainerRef>,
+        /// Whether a draining cluster instance calls its actors' `drain()` (`drain`).
+        drain: bool,
     },
     Ephemeral {
         evictable: bool,
@@ -125,6 +127,21 @@ pub enum ActorConfig {
     },
 }
 
+impl ActorConfig {
+    /// The namespace synthesized to back `workflow`, keyed `unique_key`.
+    pub fn workflow(unique_key: String, workflow: Workflow) -> Self {
+        Self::Durable {
+            unique_key,
+            // Workflow actors must not be evicted mid-run, and their state is SQL-backed.
+            evictable: false,
+            enable_sql: true,
+            workflow: Some(Box::new(workflow)),
+            container: None,
+            drain: false,
+        }
+    }
+}
+
 /// One worker's namespaces by class name, in config order.
 pub type ActorConfigMap = LinkedHashMap<String, ActorConfig>;
 
diff --git a/src/workerd/server/server/bridge.rs b/src/workerd/server/server/bridge.rs
index 20de6a1443a..cc184a9dcc9 100644
--- a/src/workerd/server/server/bridge.rs
+++ b/src/workerd/server/server/bridge.rs
@@ -171,6 +171,16 @@ pub mod ffi {
         replicated: bool,
     }
 
+    /// Whether a running actor's instance has a `drain()` method, as far as can be told without
+    /// a request on it (`actor_drain_method`).
+    #[repr(u8)]
+    enum DrainMethod {
+        Some,
+        None,
+        /// Only a request on the actor can tell: it is still being constructed, or failed to be.
+        Unknown,
+    }
+
     /// What an alarm does when another instance owns its object: runs again later, a retry
     /// that counts against no limit, or is done.
     #[repr(u8)]
@@ -393,6 +403,13 @@ pub mod ffi {
         /// handler filled in, encoded, without its name. The failures of the handler's code are
         /// in the group; an error means the handler could not be run at all.
         async fn worker_bench(worker: KjOwn<WorkerInterface>, params: &[u64]) -> Result<Vec<u64>>;
+        /// Runs the `drain()` method of the actor the request `worker` is on, if its class has
+        /// one, then waits for what it wrote to be durable. Returns whether it had one; an error
+        /// is the method's failure, which the actor's own logs report.
+        async fn worker_drain(worker: KjOwn<WorkerInterface>) -> Result<bool>;
+        /// Whether the actor's instance has a `drain()` method, looked up under the isolate lock
+        /// but outside any request, so not behind the actor's input gate.
+        async fn actor_drain_method(actor: &ActorHandle) -> Result<DrainMethod>;
         /// Constructs an actor of `class_name` whose storage lives in `storage`. `hooks` is the
         /// actor's way back to the server; `hibernation_manager` is the evicted predecessor's,
         /// whose WebSockets the actor adopts. `container` gives the actor its container's Docker
diff --git a/src/workerd/server/server/cluster-test.rs b/src/workerd/server/server/cluster-test.rs
index 4e3c3897af2..f3583369621 100644
--- a/src/workerd/server/server/cluster-test.rs
+++ b/src/workerd/server/server/cluster-test.rs
@@ -69,3 +69,10 @@ fn a_disconnected_session_is_forgotten() {
         Ok(2)
     );
 }
+
+#[test]
+fn a_drain_timeout_is_at_most_ten_seconds() {
+    assert_eq!(drain_timeout(0), Duration::ZERO);
+    assert_eq!(drain_timeout(6_000), Duration::from_secs(6));
+    assert_eq!(drain_timeout(60_000), Duration::from_secs(10));
+}
diff --git a/src/workerd/server/server/cluster.rs b/src/workerd/server/server/cluster.rs
index a692c00b46e..8c887616251 100644
--- a/src/workerd/server/server/cluster.rs
+++ b/src/workerd/server/server/cluster.rs
@@ -68,6 +68,8 @@ pub struct ClusterNode {
     nonces: RefCell<HashMap<String, u64>>,
     /// The clustered namespaces, by their part of the bucket's keys.
     namespaces: RefCell<HashMap<String, Weak<ActorNamespace>>>,
+    /// How long a drain waits for the actors' `drain()` methods (`drainTimeoutMs`).
+    pub drain_timeout: Duration,
 }
 
 /// How often an instance tidies its alarm index and adopts dead peers' alarms.
@@ -154,6 +156,21 @@ pub async fn bind(
     Ok((listener, advertise))
 }
 
+/// The longest `drainTimeoutMs` honored: a drain longer than this mostly outlasts what a
+/// supervisor waits after SIGTERM.
+const MAX_DRAIN_TIMEOUT_MS: u32 = 10_000;
+
+/// `drainTimeoutMs`, at most `MAX_DRAIN_TIMEOUT_MS`.
+fn drain_timeout(ms: u32) -> Duration {
+    if ms > MAX_DRAIN_TIMEOUT_MS {
+        tracing::warn!(
+            "cluster.drainTimeoutMs {ms} is more than {MAX_DRAIN_TIMEOUT_MS}; using \
+             {MAX_DRAIN_TIMEOUT_MS}"
+        );
+    }
+    Duration::from_millis(ms.min(MAX_DRAIN_TIMEOUT_MS).into())
+}
+
 /// Fails for an `address` that is a public IP, unless the config allows it.
 fn refuse_public(conf: cluster_config::Reader<'_>, address: &str) -> Result<()> {
     let Ok(ip) = address.parse::<std::net::SocketAddr>().map(|a| a.ip()) else {
@@ -216,6 +233,7 @@ impl ClusterNode {
             advertise: address,
             nonces: RefCell::default(),
             namespaces: RefCell::default(),
+            drain_timeout: drain_timeout(conf.get_drain_timeout_ms()),
         }))
     }
 
diff --git a/src/workerd/server/server/config-test.rs b/src/workerd/server/server/config-test.rs
index 68256665735..824c531e4a9 100644
--- a/src/workerd/server/server/config-test.rs
+++ b/src/workerd/server/server/config-test.rs
@@ -74,6 +74,7 @@ fn actor_configs_read_durable_and_ephemeral_namespaces() {
             enable_sql: true,
             workflow: None,
             container: None,
+            drain: false,
         })
     );
     assert_eq!(
diff --git a/src/workerd/server/server/config.rs b/src/workerd/server/server/config.rs
index 9be554cc1e0..8d69e1997d0 100644
--- a/src/workerd/server/server/config.rs
+++ b/src/workerd/server/server/config.rs
@@ -440,6 +440,7 @@ pub fn collect_actor_configs(
                                     service_index,
                                     namespace_index,
                                 }),
+                                drain: ns.get_drain(),
                             },
                         );
                     }
@@ -611,21 +612,12 @@ fn collect_workflow_configs(
         for error in errors {
             report.service_error(name, error);
         }
-        configs.insert(
-            key.clone(),
-            ActorConfig::Durable {
-                unique_key: key,
-                // Workflow actors must not be evicted mid-run, and their state is SQL-backed.
-                evictable: false,
-                enable_sql: true,
-                workflow: Some(Box::new(Workflow {
-                    name: workflow_name,
-                    class_name,
-                    binding_service,
-                })),
-                container: None,
-            },
-        );
+        let workflow = Workflow {
+            name: workflow_name,
+            class_name,
+            binding_service,
+        };
+        configs.insert(key.clone(), ActorConfig::workflow(key, workflow));
     }
     Ok(())
 }
diff --git a/src/workerd/server/server/run.rs b/src/workerd/server/server/run.rs
index f294fc0e7c0..b2680be92c6 100644
--- a/src/workerd/server/server/run.rs
+++ b/src/workerd/server/server/run.rs
@@ -8,6 +8,7 @@
 use std::collections::HashMap;
 use std::future::Future;
 use std::io::Write;
+use std::pin::Pin;
 use std::rc::Rc;
 use std::time::Duration;
 
@@ -24,6 +25,8 @@ use workerd_capnp::service;
 use workerd_capnp::socket;
 
 use crate::Result;
+use crate::actor::ActorNamespace;
+use crate::actor::Drained;
 use crate::bench;
 use crate::bindings::Designator;
 use crate::bindings::len_u32;
@@ -406,26 +409,15 @@ pub async fn run(
     // Draining stops the accept loops; the listeners then end once their connections have.
     match futures::future::select(serving, drain).await {
         Either::Left((result, _)) => result?,
-        Either::Right(((), serving)) => {
-            // A draining instance claims nothing more, lets its clustered objects go,
-            // `preventEviction` or not, and leaves the cluster, so peers take them over at their
-            // next listing while its connections finish. A request in flight on one could not
-            // commit anyway once a peer owns the object.
-            if let Some(node) = server.cluster() {
-                node.cluster.start_draining();
-                let workers = server
-                    .services()
-                    .filter_map(|(_, service)| service.as_worker());
-                let namespaces = workers.flat_map(|worker| worker.namespaces());
-                for (_, namespace) in namespaces.filter(|(_, ns)| ns.cluster().is_some()) {
-                    namespace.abort_all(None);
-                }
-                if let Err(e) = node.cluster.deregister().await {
-                    tracing::warn!("leaving the cluster: {e}");
-                }
-            }
+        Either::Right(((), mut serving)) => {
+            let served = match server.cluster() {
+                Some(node) => leave_cluster(&server, node, serving.as_mut()).await?,
+                None => false,
+            };
             draining.send_replace(true);
-            serving.await?;
+            if !served {
+                serving.await?;
+            }
         }
     }
 
@@ -440,6 +432,101 @@ pub async fn run(
     Ok(ffi::factory_shutdown_containers(server.factory().raw()).await?)
 }
 
+/// A draining instance claims nothing more. It keeps its lease while the running objects of the
+/// clustered namespaces that opt in flush in their `drain()` methods, so their commits are still
+/// acknowledged and no peer takes them over meanwhile; `serving` drives the listeners meanwhile.
+/// Then it lets its clustered objects go, `preventEviction` or not, and leaves the cluster, so
+/// peers take them over at their next listing while its connections finish. A request in flight
+/// on one could not commit anyway once a peer owns the object.
+///
+/// Returns whether `serving` completed. A listener that fails meanwhile is the error, once the
+/// instance has left.
+async fn leave_cluster(
+    server: &Server,
+    node: &ClusterNode,
+    mut serving: Pin<&mut impl Future<Output = Result<()>>>,
+) -> Result<bool> {
+    node.cluster.stop_claiming();
+    let workers = server
+        .services()
+        .filter_map(|(_, service)| service.as_worker());
+    let namespaces: Vec<_> = workers
+        .flat_map(|worker| worker.namespaces())
+        .filter(|(_, ns)| ns.cluster().is_some())
+        .collect();
+    let draining: Vec<_> = namespaces.iter().filter(|(_, ns)| ns.drains()).collect();
+    let mut ended = false;
+    let mut failed = None;
+    if !node.drain_timeout.is_zero() && !draining.is_empty() {
+        // Only a listener's failure ends `serving` before the drain signal.
+        let flush = std::pin::pin!(drain_actors(&draining, node.drain_timeout));
+        if let Either::Right((result, flush)) =
+            futures::future::select(flush, serving.as_mut()).await
+        {
+            ended = true;
+            match result {
+                Ok(()) => flush.await,
+                Err(e) => failed = Some(e),
+            }
+        }
+    }
+    // TODO(cluster): release each drained object's owner record (node "") so that peers claim it
+    // at once, not once their listing drops this node (up to 3 s). Not at the end of its drain:
+    // the actor still serves reads here until `abort_all`, which a peer's new epoch would make
+    // stale. And even after it, the eviction its abort starts (`Owners::evict`) still tiers the
+    // object's log into the bucket while the peer recovers from it, a race nothing yet shows is
+    // safe.
+    node.cluster.start_draining();
+    for (_, namespace) in &namespaces {
+        namespace.abort_all(None);
+    }
+    if let Err(e) = node.cluster.deregister().await {
+        tracing::warn!("leaving the cluster: {e}");
+    }
+    failed.map_or(Ok(ended), Err)
+}
+
+/// Runs the `drain()` methods of the running actors of `namespaces`, every namespace's at once,
+/// for at most `timeout`, and logs what became of each namespace's: drained, without a `drain()`,
+/// failed, cut off at the deadline, and how long it took.
+async fn drain_actors(namespaces: &[&(String, Rc<ActorNamespace>)], timeout: Duration) {
+    let drains = namespaces.iter().map(|(class, namespace)| async move {
+        let started = std::time::Instant::now();
+        let Drained {
+            drained,
+            skipped,
+            failed,
+            timed_out,
+        } = namespace.drain_all(timeout).await;
+        let ms = started.elapsed().as_millis();
+        let service = namespace.service();
+        if failed + timed_out > 0 {
+            tracing::warn!(
+                service,
+                namespace = class.as_str(),
+                drained,
+                skipped,
+                failed,
+                timed_out,
+                ms,
+                "cluster drain"
+            );
+        } else {
+            tracing::info!(
+                service,
+                namespace = class.as_str(),
+                drained,
+                skipped,
+                failed,
+                timed_out,
+                ms,
+                "cluster drain"
+            );
+        }
+    });
+    futures::future::join_all(drains).await;
+}
+
 /// The entrypoints `workerd test` or `workerd bench` runs a handler of: the channel of each,
 /// and its name ("service" or "service:entrypoint").
 type Targets = Vec<(Rc<dyn Channel>, String)>;
diff --git a/src/workerd/server/tests/cluster/cluster-test.mjs b/src/workerd/server/tests/cluster/cluster-test.mjs
index 45cc2445490..3b371cfe737 100644
--- a/src/workerd/server/tests/cluster/cluster-test.mjs
+++ b/src/workerd/server/tests/cluster/cluster-test.mjs
@@ -21,11 +21,19 @@
 //   9. Facets, their deletions and deleteAll() survive a takeover.
 
 import { spawn } from 'node:child_process';
-import { mkdir, mkdtemp, readdir, rm, writeFile } from 'node:fs/promises';
+import {
+  copyFile,
+  mkdir,
+  mkdtemp,
+  readFile,
+  readdir,
+  rm,
+  writeFile,
+} from 'node:fs/promises';
 import { tmpdir } from 'node:os';
 import { connect, createServer } from 'node:net';
 import { createHash, createHmac, randomBytes } from 'node:crypto';
-import { join } from 'node:path';
+import { dirname, join } from 'node:path';
 import { env } from 'node:process';
 import { test } from 'node:test';
 import assert from 'node:assert';
@@ -316,10 +324,27 @@ class FakeS3 {
 //   bucket     the FakeS3
 //   nodeKeys   Map from nodeId to the node's key (32-char hex), which is also
 //              what owner records name
-async function withCluster(numNodes, fn, { extraArgs = () => [] } = {}) {
+//
+// `config`, when given, edits the config's text: the edited config is written
+// next to a copy of the worker it embeds.
+async function withCluster(
+  numNodes,
+  fn,
+  { extraArgs = () => [], config = null } = {}
+) {
   // Each test run gets its own bucket and disks, so state doesn't leak
   // between tests.
   const root = await mkdtemp(join(tmpdir(), 'workerd-cluster-'));
+  let configPath = env.WD_TEST_CONFIG;
+  if (config !== null) {
+    configPath = join(root, 'cluster-test.wd-test');
+    const text = await readFile(env.WD_TEST_CONFIG, 'utf8');
+    await writeFile(configPath, config(text));
+    await copyFile(
+      join(dirname(env.WD_TEST_CONFIG), 'cluster-worker.js'),
+      join(root, 'cluster-worker.js')
+    );
+  }
   const bucket = new FakeS3();
   const nodes = [];
   const nodeKeys = new Map();
@@ -330,7 +355,7 @@ async function withCluster(numNodes, fn, { extraArgs = () => [] } = {}) {
       await mkdir(localPath);
       const node = new ClusterNode({
         binary: env.WORKERD_BINARY,
-        config: env.WD_TEST_CONFIG,
+        config: configPath,
         localPath,
         s3Port: bucket.port,
         nodeId: `node${i}`,
@@ -658,6 +683,206 @@ test(
   }
 );
 
+// `drainTimeoutMs` in cluster-test.wd-test.
+const DRAIN_TIMEOUT_MS = 3000;
+
+// The response to `path` from `node` once it serves it, within `withinMs`:
+// a peer routes to a departed owner until its listing drops it.
+async function fetchOnceServed(node, path, withinMs = 10_000) {
+  const deadline = Date.now() + withinMs;
+  let response = null;
+  while (Date.now() < deadline) {
+    response = await fetchJson(node.httpPort, path, {
+      timeoutMs: Math.max(deadline - Date.now(), 1),
+    }).catch(() => null);
+    if (response?.status === 200) return response;
+    await sleep(200);
+  }
+  assert.fail(`${path} not served: ${JSON.stringify(response?.body)}`);
+}
+
+test(
+  "cluster: a draining owner's DO flushes in drain(), and the next owner sees it",
+  SIGTERM_DRAINS,
+  async () => {
+    await withCluster(2, async ({ nodes }) => {
+      const first = await fetchJson(nodes[0].httpPort, '/increment?name=flush');
+      assert.strictEqual(first.status, 200, JSON.stringify(first.body));
+      const owner = nodes.find((n) => n.nodeId === first.body.nodeId);
+      const peer = nodes.find((n) => n.nodeId !== first.body.nodeId);
+      for (let i = 1; i <= 3; i++) {
+        const buffered = await fetchJson(owner.httpPort, '/buffer?name=flush');
+        assert.strictEqual(buffered.body.buffered, i);
+      }
+
+      // Kept in memory only, until the owner drains.
+      const exit = await owner.stop({ signal: 'SIGTERM', timeoutMs: 20_000 });
+      assert.deepStrictEqual(exit, { code: 0, signal: null });
+      assert.match(
+        owner.stderr,
+        /cluster drain; service = "main"; namespace = "Counter"; drained = 1;/
+      );
+      assert.match(owner.stderr, /namespace = "Counter";.*; timed_out = 0;/);
+
+      const after = await fetchOnceServed(peer, '/get?name=flush');
+      assert.strictEqual(after.body.nodeId, peer.nodeId);
+      assert.strictEqual(after.body.count, 1);
+      assert.strictEqual(after.body.buffered, 3);
+    });
+  }
+);
+
+test(
+  'cluster: a drain() that does not finish is cut off at the deadline',
+  SIGTERM_DRAINS,
+  async () => {
+    await withCluster(2, async ({ nodes }) => {
+      const first = await fetchJson(nodes[0].httpPort, '/increment?name=hang');
+      assert.strictEqual(first.status, 200, JSON.stringify(first.body));
+      const owner = nodes.find((n) => n.nodeId === first.body.nodeId);
+      const peer = nodes.find((n) => n.nodeId !== first.body.nodeId);
+      const buffered = await fetchJson(owner.httpPort, '/buffer?name=hang');
+      assert.strictEqual(buffered.status, 200);
+      const hung = await fetchJson(owner.httpPort, '/hang-drain?name=hang');
+      assert.strictEqual(hung.status, 200);
+
+      const signalled = Date.now();
+      const exit = await owner.stop({ signal: 'SIGTERM', timeoutMs: 20_000 });
+      const elapsed = Date.now() - signalled;
+      assert.deepStrictEqual(exit, { code: 0, signal: null });
+      assert(
+        elapsed >= DRAIN_TIMEOUT_MS - 100 && elapsed < DRAIN_TIMEOUT_MS + 5000,
+        `exited ${elapsed} ms after SIGTERM`
+      );
+      assert.match(owner.stderr, /namespace = "Counter";.*; timed_out = 1;/);
+
+      // What drain() wrote before it hung, committed long before the deadline
+      // while the owner kept its lease, is the next owner's.
+      const after = await fetchOnceServed(peer, '/get?name=hang');
+      assert.strictEqual(after.body.nodeId, peer.nodeId);
+      assert.strictEqual(after.body.count, 1);
+      assert.strictEqual(after.body.buffered, 1);
+    });
+  }
+);
+
+// SIGTERMs `owner`, which must exit cleanly; returns how long it took, in ms.
+async function drainOwner(owner) {
+  const signalled = Date.now();
+  const exit = await owner.stop({ signal: 'SIGTERM', timeoutMs: 20_000 });
+  assert.deepStrictEqual(exit, { code: 0, signal: null });
+  return Date.now() - signalled;
+}
+
+// Starts the DO `name` (of namespace `ns`, the Counter by default) on whichever
+// node claims it; returns that node and the other.
+async function ownerAndPeer(nodes, name, ns) {
+  const query = ns === undefined ? '' : `&ns=${ns}`;
+  const first = await fetchJson(
+    nodes[0].httpPort,
+    `/identity?name=${name}${query}`
+  );
+  assert.strictEqual(first.status, 200, JSON.stringify(first.body));
+  return [
+    nodes.find((n) => n.nodeId === first.body.nodeId),
+    nodes.find((n) => n.nodeId !== first.body.nodeId),
+  ];
+}
+
+test(
+  'cluster: a namespace that does not opt in is not drained',
+  SIGTERM_DRAINS,
+  async () => {
+    await withCluster(2, async ({ nodes }) => {
+      const [owner, peer] = await ownerAndPeer(nodes, 'unopted', 'unopted');
+      const buffered = await fetchJson(
+        owner.httpPort,
+        '/buffer?name=unopted&ns=unopted'
+      );
+      assert.strictEqual(buffered.body.buffered, 1);
+      await drainOwner(owner);
+      assert.doesNotMatch(owner.stderr, /namespace = "Unopted"/);
+
+      // drain() never ran: what it would have written is lost.
+      const after = await fetchOnceServed(peer, '/get?name=unopted&ns=unopted');
+      assert.strictEqual(after.body.nodeId, peer.nodeId);
+      assert.strictEqual(after.body.buffered, 0);
+    });
+  }
+);
+
+test(
+  'cluster: an opted-in DO whose class has no drain() is skipped',
+  SIGTERM_DRAINS,
+  async () => {
+    await withCluster(2, async ({ nodes }) => {
+      const [owner] = await ownerAndPeer(nodes, 'plain', 'no-drain');
+      const elapsed = await drainOwner(owner);
+      const line = owner.stderr.match(
+        /namespace = "NoDrain"; drained = 0; skipped = 1; failed = 0; timed_out = 0; ms = (\d+)/
+      );
+      assert(line !== null, 'no drain line for NoDrain');
+      assert(Number(line[1]) < 1000, `skipping took ${line[1]} ms`);
+      assert(elapsed < DRAIN_TIMEOUT_MS, `exited ${elapsed} ms after SIGTERM`);
+    });
+  }
+);
+
+test(
+  'cluster: a drain() that throws fails alone, and the node still exits',
+  SIGTERM_DRAINS,
+  async () => {
+    await withCluster(2, async ({ nodes }) => {
+      const [owner, peer] = await ownerAndPeer(nodes, 'throws');
+      await fetchJson(owner.httpPort, '/buffer?name=throws');
+      await fetchJson(owner.httpPort, '/throw-drain?name=throws');
+      await drainOwner(owner);
+      assert.match(
+        owner.stderr,
+        /namespace = "Counter"; drained = 0; skipped = 0; failed = 1; timed_out = 0;/
+      );
+
+      // It wrote before it threw.
+      const after = await fetchOnceServed(peer, '/get?name=throws');
+      assert.strictEqual(after.body.nodeId, peer.nodeId);
+      assert.strictEqual(after.body.buffered, 1);
+    });
+  }
+);
+
+test(
+  'cluster: drainTimeoutMs = 0 calls no drain()',
+  SIGTERM_DRAINS,
+  async () => {
+    const config = (text) => {
+      const edited = text.replace(
+        'drainTimeoutMs = 3000',
+        'drainTimeoutMs = 0'
+      );
+      assert.notStrictEqual(edited, text);
+      return edited;
+    };
+    await withCluster(
+      2,
+      async ({ nodes }) => {
+        const [owner, peer] = await ownerAndPeer(nodes, 'off');
+        await fetchJson(owner.httpPort, '/buffer?name=off');
+        await fetchJson(owner.httpPort, '/hang-drain?name=off');
+        const elapsed = await drainOwner(owner);
+        assert(
+          elapsed < DRAIN_TIMEOUT_MS,
+          `exited ${elapsed} ms after SIGTERM`
+        );
+        assert.doesNotMatch(owner.stderr, /cluster drain/);
+
+        const after = await fetchOnceServed(peer, '/get?name=off');
+        assert.strictEqual(after.body.buffered, 0);
+      },
+      { config }
+    );
+  }
+);
+
 test('cluster: concurrent first requests to a DO read its owner record once per node', async () => {
   await withCluster(2, async ({ nodes, bucket }) => {
     // Without coalescing, each of these reads own.json and the losers retry
diff --git a/src/workerd/server/tests/cluster/cluster-test.wd-test b/src/workerd/server/tests/cluster/cluster-test.wd-test
index 147a47732a2..c8ddeb8327c 100644
--- a/src/workerd/server/tests/cluster/cluster-test.wd-test
+++ b/src/workerd/server/tests/cluster/cluster-test.wd-test
@@ -26,10 +26,16 @@ const config :Workerd.Config = (
         compatibilityFlags = ["nodejs_compat", "allow_irrevocable_stub_storage"],
         bindings = [
           ( name = "COUNTER", durableObjectNamespace = "Counter" ),
+          ( name = "NO_DRAIN", durableObjectNamespace = "NoDrain" ),
+          ( name = "UNOPTED", durableObjectNamespace = "Unopted" ),
           ( name = "NODE_ID", fromEnvironment = "NODE_ID" ),
         ],
         durableObjectNamespaces = [
-          ( className = "Counter", uniqueKey = "counter-test-namespace" ),
+          ( className = "Counter", uniqueKey = "counter-test-namespace", drain = true ),
+          # Opted in, with a class that has no drain().
+          ( className = "NoDrain", uniqueKey = "no-drain-test-namespace", drain = true ),
+          # A Counter, with its drain(), in a namespace that does not opt in.
+          ( className = "Unopted", uniqueKey = "unopted-test-namespace" ),
         ],
         durableObjectStorage = ( localDisk = "local" ),
       )
@@ -49,5 +55,7 @@ const config :Workerd.Config = (
     ],
     address = "127.0.0.1:0",
     key = "test-cluster-shared-secret-of-32-bytes",
+    # Shorter than the default, so that a drain cut off at it is quick to test.
+    drainTimeoutMs = 3000,
   ),
 );
diff --git a/src/workerd/server/tests/cluster/cluster-worker.js b/src/workerd/server/tests/cluster/cluster-worker.js
index ce8d20925dc..35db238b3d4 100644
--- a/src/workerd/server/tests/cluster/cluster-worker.js
+++ b/src/workerd/server/tests/cluster/cluster-worker.js
@@ -69,6 +69,15 @@
 //   /set-go          Sets this DO's `go` flag (read by Consumer via /get).
 //   /record?value=V  Stores V (JSON) as this DO's `recorded` value.
 //
+// Drain paths (see drain()):
+//   /buffer          Adds 1 to a count this instance keeps in memory only, which
+//                    its drain() adds to `buffered` in storage. /get returns
+//                    `buffered`.
+//   /hang-drain      Makes this instance's drain() never finish.
+//   /throw-drain     Makes this instance's drain() throw.
+// The top-level fetch() sends a request to the NoDrain or Unopted DO of the
+// name instead when given `ns=no-drain` or `ns=unopted`.
+//
 // The top-level fetch() additionally honors a `delay` query parameter (ms): it
 // waits that long *before* calling the DO. This lets a test put a request in
 // flight on a node and SIGTERM that node while the request has not yet routed
@@ -91,8 +100,12 @@ export default {
     const name = url.searchParams.get('name') ?? 'default';
 
     // Synthesize a stub URL for the DO.
-    const id = env.COUNTER.idFromName(name);
-    const stub = env.COUNTER.get(id);
+    const ns =
+      { 'no-drain': env.NO_DRAIN, unopted: env.UNOPTED }[
+        url.searchParams.get('ns')
+      ] ?? env.COUNTER;
+    const id = ns.idFromName(name);
+    const stub = ns.get(id);
 
     const delay = url.searchParams.get('delay');
     if (delay !== null) {
@@ -193,6 +206,10 @@ export class Counter extends DurableObject {
     this.env = env;
     // See /hold-stub. Maps target DO name -> stub.
     this.heldStubs = new Map();
+    // See /buffer and /hang-drain.
+    this.buffered = 0;
+    this.hangDrain = false;
+    this.throwDrain = false;
 
     state.blockConcurrencyWhile(async () => {
       if (await state.storage.get(BREAK_ON_CONSTRUCT)) {
@@ -224,11 +241,13 @@ export class Counter extends DurableObject {
       const restores = (await this.state.storage.get('restores')) ?? 0;
       const go = (await this.state.storage.get('go')) ?? false;
       const recorded = await this.state.storage.get('recorded');
+      const buffered = (await this.state.storage.get('buffered')) ?? 0;
       return Response.json({
         count,
         restores,
         go,
         recorded,
+        buffered,
         nodeId,
         id: idHex,
       });
@@ -312,6 +331,15 @@ export class Counter extends DurableObject {
         JSON.parse(url.searchParams.get('value'))
       );
       return Response.json({ nodeId, id: idHex });
+    } else if (url.pathname === '/buffer') {
+      this.buffered += 1;
+      return Response.json({ nodeId, id: idHex, buffered: this.buffered });
+    } else if (url.pathname === '/hang-drain') {
+      this.hangDrain = true;
+      return Response.json({ nodeId, id: idHex });
+    } else if (url.pathname === '/throw-drain') {
+      this.throwDrain = true;
+      return Response.json({ nodeId, id: idHex });
     } else if (url.pathname === '/hold') {
       await scheduler.wait(1000);
       return Response.json({ nodeId, id: idHex });
@@ -358,6 +386,22 @@ export class Counter extends DurableObject {
     await this.state.storage.put('fired', fired);
   }
 
+  // Called once as this instance drains (SIGTERM), while it still owns the DO:
+  // writes what /buffer kept in memory, and after /hang-drain never finishes.
+  async drain() {
+    if (this.buffered > 0) {
+      const buffered = (await this.state.storage.get('buffered')) ?? 0;
+      await this.state.storage.put('buffered', buffered + this.buffered);
+      this.buffered = 0;
+    }
+    if (this.hangDrain) {
+      await new Promise(() => {});
+    }
+    if (this.throwDrain) {
+      throw new Error('drain() failed on purpose');
+    }
+  }
+
   // RPC method used by /vend-stub on the *target* DO.
   vend(kind) {
     return this.state.restore({ kind });
@@ -371,3 +415,17 @@ export class Counter extends DurableObject {
     throw new Error(`unexpected restore params: ${JSON.stringify(params)}`);
   }
 }
+
+// A Counter whose namespace does not opt in to drain(): its drain() is never
+// called.
+export class Unopted extends Counter {}
+
+// A DO of a namespace that opts in to drain(), whose class has none.
+export class NoDrain extends DurableObject {
+  fetch() {
+    return Response.json({
+      nodeId: this.env.NODE_ID ?? '<unknown>',
+      id: this.ctx.id.toString(),
+    });
+  }
+}
diff --git a/src/workerd/server/workerd.capnp b/src/workerd/server/workerd.capnp
index 63457f6613c..aeb6b23803d 100644
--- a/src/workerd/server/workerd.capnp
+++ b/src/workerd/server/workerd.capnp
@@ -156,6 +156,16 @@ struct ClusterConfig {
   # Allow a public IP as `address` or `advertise`. Refused by default, since traffic between
   # instances is not encrypted. Private, loopback, link-local and shared (100.64.0.0/10, as
   # Tailscale uses) addresses, wildcards and hostnames are allowed.
+
+  drainTimeoutMs @6 :UInt32 = 6000;
+  # How long a draining instance (SIGTERM) gives its running Durable Objects to flush before it
+  # lets them go. It first stops claiming objects, keeping its lease, so the objects it owns stay
+  # its own and their writes are still acknowledged; then calls the `drain()` method of each
+  # running Durable Object of the namespaces that opt in (`DurableObjectNamespace.drain`) whose
+  # class defines one, and waits for them, and for what they wrote to be durable, for at most this
+  # long. Then it lets its objects go and leaves the cluster. 0 skips the `drain()` calls; more
+  # than 10000 is taken as 10000. Keep it well within the time the instance has to exit after
+  # SIGTERM.
 }
 
 struct LoggingOptions {
@@ -792,6 +802,13 @@ struct Worker {
     # emulate behavior of traditional DO namespaces on Cloudflare that aren't SQLite-backed. This
     # flag should be enabled when testing code that will run on a SQLite-backed namespace.
 
+    drain @6 :Bool = false;
+    # In cluster mode (`Config.cluster`), call the `drain()` method of each of this namespace's
+    # Durable Objects running on an instance as that instance drains (SIGTERM), while it still owns
+    # them, so that they can write out what they keep in memory: see `ClusterConfig.drainTimeoutMs`.
+    # Off by default, so that a class's own method named `drain` is not called by surprise. Objects
+    # whose class has no `drain()` method are skipped.
+
     container @5 :ContainerOptions;
     # If present, Durable Objects in this namespace have attached containers.
     # workerd will talk to the configured container engine to start containers for each Durable