From 696f42bab0761b73337110ad1f4bd1ddca63413f Mon Sep 17 00:00:00 2001 From: Gogo Date: Tue, 10 Nov 2026 07:39:06 +0000 Subject: [PATCH 1/7] cluster: Cluster::stop_claiming, which keeps the lease A draining instance stops claiming objects but, unlike start_draining, keeps renewing its lease: the objects it owns stay its own, their commits are still acknowledged, and no peer can claim them while they flush. Signed-off-by: Gogo Co-Authored-By: Claude Opus 5.5 --- src/workerd/server/cluster/cluster.rs | 54 +++++++++++++++++++++++++++ 1 file changed, 54 insertions(+) diff --git a/src/workerd/server/cluster/cluster.rs b/src/workerd/server/cluster/cluster.rs index 46ee4a9c4b7..b1081137898 100644 --- a/src/workerd/server/cluster/cluster.rs +++ b/src/workerd/server/cluster/cluster.rs @@ -388,6 +388,13 @@ 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`. + 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) { @@ -450,12 +457,59 @@ async fn check_conditional_writes(bucket: &Bucket, node: &str) -> Result<()> { #[cfg(test)] mod tests { + 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; + fn node(bucket: &Bucket, name: &str) -> Rc { + 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"), + } + + a.start_draining(); + assert!(a.lease.borrow().draining); + }); + } + #[test] fn a_bucket_that_ignores_preconditions_is_refused() { block_on(async { From 27a2dc00565231dd1fe726bff8de78609f51f50f Mon Sep 17 00:00:00 2001 From: Gogo Date: Tue, 10 Nov 2026 07:39:06 +0000 Subject: [PATCH 2/7] server: call clustered Durable Objects' drain() before letting them go On SIGTERM a cluster instance now first stops claiming (keeping its lease), then calls the drain() method of every running clustered Durable Object whose class defines one, all at once, and waits for them and for what they wrote to be durable, for at most cluster.drainTimeoutMs (default 6000; 0 skips it). The listeners keep serving meanwhile. Then it lets its objects go and leaves the cluster as before (start_draining, abort_all, deregister). drain() is a custom event on the actor's own request, started the way its alarms and hibernated WebSocket events are (ActorContainer's start_request); the factory's DrainCustomEvent looks the method up on the instance, as [restore]() is, so classes without one are skipped. Each namespace logs how many actors drained, had no drain(), failed and were cut off, and how long it took. Signed-off-by: Gogo Co-Authored-By: Claude Opus 5.5 --- .../server/factory/worker-factory-actor.c++ | 81 ++++++++++++++++++ src/workerd/server/factory/worker-factory.h | 1 + src/workerd/server/server/actor.rs | 56 ++++++++++++ src/workerd/server/server/bridge.rs | 4 + src/workerd/server/server/cluster.rs | 3 + src/workerd/server/server/run.rs | 85 +++++++++++++++++-- src/workerd/server/workerd.capnp | 9 ++ 7 files changed, 230 insertions(+), 9 deletions(-) diff --git a/src/workerd/server/factory/worker-factory-actor.c++ b/src/workerd/server/factory/worker-factory-actor.c++ index c0a38f7ab5e..234dcebc1d0 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 +#include #include #include #include @@ -458,6 +459,79 @@ 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 run(kj::Own incomingRequest, + kj::Maybe entrypointName, + kj::Maybe 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 { + 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()) { + ran = true; + v8::LocalVector args(js.v8Isolate); + return ioctx.awaitJs(js, js.tryCatch([&] { + return js.toVoidPromise(method.call(js, self, args)); + }, [&](jsg::Value exception) { return js.rejectedPromise(kj::mv(exception)); })); + } + } + return kj::READY_NOW; + }); + 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 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 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 +847,13 @@ kj::Maybe> actor_hibernation_manager(const ActorHandl [](HibernationManager& m) { return kj::Rc(m.addRef()); }); } +kj::Promise worker_drain(kj::Own worker) { + bool ran = false; + auto result = co_await worker->customEvent(kj::heap(ran)); + KJ_REQUIRE(result.outcome == EventOutcome::OK, "the Durable Object's drain() failed"); + co_return ran; +} + kj::Promise facet_start_resolve(kj::Own 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..760450f54bb 100644 --- a/src/workerd/server/factory/worker-factory.h +++ b/src/workerd/server/factory/worker-factory.h @@ -217,6 +217,7 @@ kj::Own worker_interface_attach( kj::Own inner, ::rust::Box keep); kj::Promise<::rust::Vec> worker_bench( kj::Own worker, ::rust::Slice params); +kj::Promise worker_drain(kj::Own worker); kj::Own worker_new_actor(const CompiledWorker& worker, ::rust::Str className, kj::Maybe> props, diff --git a/src/workerd/server/server/actor.rs b/src/workerd/server/server/actor.rs index 20f4d74666f..aef8ecf5f16 100644 --- a/src/workerd/server/server/actor.rs +++ b/src/workerd/server/server/actor.rs @@ -109,6 +109,19 @@ pub struct ActorNamespace { resolving: RefCell>>, } +/// 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>>, @@ -373,6 +386,42 @@ impl ActorNamespace { } } + /// Runs the `drain()` method of every running root actor whose class has one, all at once, + /// cutting 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, factory) = (Rc::clone(container), Rc::clone(&self.factory)); + async move { + let drain = std::pin::pin!(container.drain()); + let deadline = std::pin::pin!(factory.sleep(timeout)); + match futures::future::select(drain, deadline).await { + Either::Left((result, _)) => Some(result), + // Dropping the drain cancels the method's request. + Either::Right(_) => None, + } + } + }) + .collect(); + let mut drained = Drained::default(); + for result in futures::future::join_all(drains).await { + match result { + Some(Ok(true)) => drained.drained += 1, + Some(Ok(false)) => drained.skipped += 1, + Some(Err(_)) => drained.failed += 1, + None => drained.timed_out += 1, + } + } + 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 +1051,13 @@ impl ActorContainer { Ok(()) } + /// Runs the actor's `drain()` method as a request of its own: whether its class has one. It + /// completes once what the method wrote is durable. + async fn drain(self: &Rc) -> Result { + 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/bridge.rs b/src/workerd/server/server/bridge.rs index 20de6a1443a..9484e190a2e 100644 --- a/src/workerd/server/server/bridge.rs +++ b/src/workerd/server/server/bridge.rs @@ -393,6 +393,10 @@ 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, params: &[u64]) -> Result>; + /// 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) -> Result; /// 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.rs b/src/workerd/server/server/cluster.rs index a692c00b46e..fd188a40421 100644 --- a/src/workerd/server/server/cluster.rs +++ b/src/workerd/server/server/cluster.rs @@ -68,6 +68,8 @@ pub struct ClusterNode { nonces: RefCell>, /// The clustered namespaces, by their part of the bucket's keys. namespaces: RefCell>>, + /// 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. @@ -216,6 +218,7 @@ impl ClusterNode { advertise: address, nonces: RefCell::default(), namespaces: RefCell::default(), + drain_timeout: Duration::from_millis(conf.get_drain_timeout_ms().into()), })) } diff --git a/src/workerd/server/server/run.rs b/src/workerd/server/server/run.rs index f294fc0e7c0..8e9b64113d5 100644 --- a/src/workerd/server/server/run.rs +++ b/src/workerd/server/server/run.rs @@ -24,6 +24,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,18 +408,43 @@ 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. + Either::Right(((), mut serving)) => { + // A draining instance claims nothing more. It keeps its lease while its running + // clustered objects' `drain()` methods flush them, so their commits are still + // acknowledged and no peer takes them over 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. + let mut served = false; if let Some(node) = server.cluster() { - node.cluster.start_draining(); + node.cluster.stop_claiming(); 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()) { + let namespaces: Vec<_> = workers + .flat_map(|worker| worker.namespaces()) + .filter(|(_, ns)| ns.cluster().is_some()) + .collect(); + if !node.drain_timeout.is_zero() { + // The listeners serve on meanwhile: their connections are driven by + // `serving`. Only a listener's failure ends it before the drain signal. + let flush = std::pin::pin!(drain_actors(&namespaces, node.drain_timeout)); + if let Either::Right((result, flush)) = + futures::future::select(flush, serving.as_mut()).await + { + result?; + served = true; + flush.await; + } + } + // 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 { @@ -425,7 +452,9 @@ pub async fn run( } } draining.send_replace(true); - serving.await?; + if !served { + serving.await?; + } } } @@ -440,6 +469,44 @@ pub async fn run( Ok(ffi::factory_shutdown_containers(server.factory().raw()).await?) } +/// 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)], 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(); + if failed + timed_out > 0 { + tracing::warn!( + namespace = class.as_str(), + drained, + skipped, + failed, + timed_out, + ms, + "cluster drain" + ); + } else { + tracing::info!( + 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, String)>; diff --git a/src/workerd/server/workerd.capnp b/src/workerd/server/workerd.capnp index 63457f6613c..ce408783050 100644 --- a/src/workerd/server/workerd.capnp +++ b/src/workerd/server/workerd.capnp @@ -156,6 +156,15 @@ 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 whose class defines one, all at once, 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, as it does without a `drain()`. 0 skips the `drain()` calls. Keep it well within the + # time the instance has to exit after SIGTERM, and under the lease (20 s). } struct LoggingOptions { From 1b3741690f2491f5a923a82fe2b800756839a2dc Mon Sep 17 00:00:00 2001 From: Gogo Date: Tue, 10 Nov 2026 07:39:06 +0000 Subject: [PATCH 3/7] Cluster test: drain() flushes for the next owner; a hung one is cut off Signed-off-by: Gogo Co-Authored-By: Claude Opus 5.5 --- .../server/tests/cluster/cluster-test.mjs | 80 +++++++++++++++++++ .../server/tests/cluster/cluster-test.wd-test | 2 + .../server/tests/cluster/cluster-worker.js | 30 +++++++ 3 files changed, 112 insertions(+) diff --git a/src/workerd/server/tests/cluster/cluster-test.mjs b/src/workerd/server/tests/cluster/cluster-test.mjs index 45cc2445490..80be4f62e73 100644 --- a/src/workerd/server/tests/cluster/cluster-test.mjs +++ b/src/workerd/server/tests/cluster/cluster-test.mjs @@ -658,6 +658,86 @@ 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.*drained = 1/); + assert.match(owner.stderr, /cluster drain.*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, /cluster drain.*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); + }); + } +); + 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..4eba21565b1 100644 --- a/src/workerd/server/tests/cluster/cluster-test.wd-test +++ b/src/workerd/server/tests/cluster/cluster-test.wd-test @@ -49,5 +49,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..f829c8b8241 100644 --- a/src/workerd/server/tests/cluster/cluster-worker.js +++ b/src/workerd/server/tests/cluster/cluster-worker.js @@ -69,6 +69,12 @@ // /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. +// // 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 @@ -193,6 +199,9 @@ 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; state.blockConcurrencyWhile(async () => { if (await state.storage.get(BREAK_ON_CONSTRUCT)) { @@ -224,11 +233,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 +323,12 @@ 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 === '/hold') { await scheduler.wait(1000); return Response.json({ nodeId, id: idHex }); @@ -358,6 +375,19 @@ 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(() => {}); + } + } + // RPC method used by /vend-stub on the *target* DO. vend(kind) { return this.state.restore({ kind }); From d5482e9a7dbec795495ac8fdb78d0af27806c6c6 Mon Sep 17 00:00:00 2001 From: Gogo Date: Tue, 10 Nov 2026 08:06:35 +0000 Subject: [PATCH 4/7] cluster: a node that stopped claiming still wakes what it owns While only stop_claiming is in effect, an owner record naming this node is taken up here (Owned), so an object it owns but does not have open wakes locally while its lease holds; it is just not drained. Claiming an unowned object, or one whose owner died, is still refused, with the existing error the server returns as DISCONNECTED. start_draining refuses both, as before. Signed-off-by: Gogo Co-Authored-By: Claude Opus 5.5 --- src/workerd/server/cluster/cluster.rs | 11 +++++++++-- src/workerd/server/cluster/ownership.rs | 9 +++++++-- 2 files changed, 16 insertions(+), 4 deletions(-) diff --git a/src/workerd/server/cluster/cluster.rs b/src/workerd/server/cluster/cluster.rs index b1081137898..3ba399684b2 100644 --- a/src/workerd/server/cluster/cluster.rs +++ b/src/workerd/server/cluster/cluster.rs @@ -390,7 +390,8 @@ 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`. + /// 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); } @@ -400,6 +401,7 @@ impl Cluster { 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. @@ -499,14 +501,19 @@ mod tests { 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. + // 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()); }); } diff --git a/src/workerd/server/cluster/ownership.rs b/src/workerd/server/cluster/ownership.rs index ab0051084d9..3b6731fb06e 100644 --- a/src/workerd/server/cluster/ownership.rs +++ b/src/workerd/server/cluster/ownership.rs @@ -79,6 +79,9 @@ pub struct Ownership { registry: Rc, /// Whether this instance may claim objects: off once it drains. pub may_claim: Cell, + /// 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, /// Where each routed object's owner was, and when that was read. routes: RefCell>, /// The record each claim wrote, and the guard that it is still the @@ -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())); From 0084b4300e55f40f6c27a5f19928d5649c020f15 Mon Sep 17 00:00:00 2001 From: Gogo Date: Tue, 10 Nov 2026 08:06:35 +0000 Subject: [PATCH 5/7] server: drain() is opt-in per Durable Object namespace - DurableObjectNamespace.drain (default false): only the namespaces that opt in are drained, so a class's own method named drain is never called by surprise. ClusterConfig.drainTimeoutMs stays the node-wide deadline, now at most 10000 (a larger value is logged and clamped). - An actor's drain() method is looked up under the isolate lock but outside any request (actor_drain_method), so an actor without one is skipped without waiting behind its input gate; the event waits for output locks only when the method ran. - At most 32 actors of a namespace drain at once, under one deadline. - A listener failing during the drain no longer skips start_draining, abort_all and deregister: the instance leaves the cluster, then returns the error (leave_cluster in run.rs). - The per-namespace log line names the service too. Signed-off-by: Gogo Co-Authored-By: Claude Opus 5.5 --- .../server/factory/worker-factory-actor.c++ | 30 ++++- src/workerd/server/factory/worker-factory.h | 2 + src/workerd/server/server/actor.rs | 63 +++++++---- src/workerd/server/server/bindings-test.rs | 1 + src/workerd/server/server/bindings.rs | 17 +++ src/workerd/server/server/bridge.rs | 13 +++ src/workerd/server/server/cluster-test.rs | 7 ++ src/workerd/server/server/cluster.rs | 17 ++- src/workerd/server/server/config-test.rs | 1 + src/workerd/server/server/config.rs | 22 ++-- src/workerd/server/server/run.rs | 106 +++++++++++------- src/workerd/server/workerd.capnp | 16 ++- 12 files changed, 210 insertions(+), 85 deletions(-) diff --git a/src/workerd/server/factory/worker-factory-actor.c++ b/src/workerd/server/factory/worker-factory-actor.c++ index 234dcebc1d0..99efd7ff897 100644 --- a/src/workerd/server/factory/worker-factory-actor.c++ +++ b/src/workerd/server/factory/worker-factory-actor.c++ @@ -497,7 +497,8 @@ class DrainCustomEvent final: public WorkerInterface::CustomEvent { } return kj::READY_NOW; }); - co_await context.waitForOutputLocks(); + // 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( @@ -847,6 +848,33 @@ kj::Maybe> actor_hibernation_manager(const ActorHandl [](HibernationManager& m) { return kj::Rc(m.addRef()); }); } +kj::Promise 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)) { + auto self = jsg::JsObject(handler->self.getHandle(js)); + // A getter or a Proxy trap that throws leaves it to the request. + method = js.tryCatch([&] { + return self.get(js, "drain"_kj).tryCast() == kj::none + ? DrainMethod::None + : DrainMethod::Some; + }, [&](jsg::Value) { return DrainMethod::Unknown; }); + } else { + method = DrainMethod::None; + } + }); + }); + }); + co_return method; +} + kj::Promise worker_drain(kj::Own worker) { bool ran = false; auto result = co_await worker->customEvent(kj::heap(ran)); diff --git a/src/workerd/server/factory/worker-factory.h b/src/workerd/server/factory/worker-factory.h index 760450f54bb..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}; @@ -218,6 +219,7 @@ kj::Own worker_interface_attach( kj::Promise<::rust::Vec> worker_bench( kj::Own worker, ::rust::Slice params); kj::Promise worker_drain(kj::Own worker); +kj::Promise actor_drain_method(const ActorHandle& actor); kj::Own worker_new_actor(const CompiledWorker& worker, ::rust::Str className, kj::Maybe> props, diff --git a/src/workerd/server/server/actor.rs b/src/workerd/server/server/actor.rs index aef8ecf5f16..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 @@ -386,9 +389,20 @@ impl ActorNamespace { } } - /// Runs the `drain()` method of every running root actor whose class has one, all at once, - /// cutting off those not done `timeout` from now. The actors keep serving meanwhile, and stay - /// once drained: the caller lets them go (`abort_all`). + /// 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. @@ -398,25 +412,25 @@ impl ActorNamespace { .values() .filter(|container| container.is_running() && container.require_not_broken().is_ok()) .map(|container| { - let (container, factory) = (Rc::clone(container), Rc::clone(&self.factory)); - async move { - let drain = std::pin::pin!(container.drain()); - let deadline = std::pin::pin!(factory.sleep(timeout)); - match futures::future::select(drain, deadline).await { - Either::Left((result, _)) => Some(result), - // Dropping the drain cancels the method's request. - Either::Right(_) => None, - } - } + 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(); - for result in futures::future::join_all(drains).await { - match result { - Some(Ok(true)) => drained.drained += 1, - Some(Ok(false)) => drained.skipped += 1, - Some(Err(_)) => drained.failed += 1, - None => drained.timed_out += 1, + 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 @@ -1051,9 +1065,16 @@ impl ActorContainer { Ok(()) } - /// Runs the actor's `drain()` method as a request of its own: whether its class has one. It - /// completes once what the method wrote is durable. + /// 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) -> Result { + 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?) } 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>, /// Where the config has the class's `container` options, if it has any. container: Option, + /// 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; diff --git a/src/workerd/server/server/bridge.rs b/src/workerd/server/server/bridge.rs index 9484e190a2e..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)] @@ -397,6 +407,9 @@ pub mod ffi { /// 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) -> Result; + /// 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; /// 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 fd188a40421..8c887616251 100644 --- a/src/workerd/server/server/cluster.rs +++ b/src/workerd/server/server/cluster.rs @@ -156,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::().map(|a| a.ip()) else { @@ -218,7 +233,7 @@ impl ClusterNode { advertise: address, nonces: RefCell::default(), namespaces: RefCell::default(), - drain_timeout: Duration::from_millis(conf.get_drain_timeout_ms().into()), + 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 8e9b64113d5..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; @@ -409,48 +410,10 @@ pub async fn run( match futures::future::select(serving, drain).await { Either::Left((result, _)) => result?, Either::Right(((), mut serving)) => { - // A draining instance claims nothing more. It keeps its lease while its running - // clustered objects' `drain()` methods flush them, so their commits are still - // acknowledged and no peer takes them over 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. - let mut served = false; - if let Some(node) = server.cluster() { - 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(); - if !node.drain_timeout.is_zero() { - // The listeners serve on meanwhile: their connections are driven by - // `serving`. Only a listener's failure ends it before the drain signal. - let flush = std::pin::pin!(drain_actors(&namespaces, node.drain_timeout)); - if let Either::Right((result, flush)) = - futures::future::select(flush, serving.as_mut()).await - { - result?; - served = true; - flush.await; - } - } - // 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}"); - } - } + let served = match server.cluster() { + Some(node) => leave_cluster(&server, node, serving.as_mut()).await?, + None => false, + }; draining.send_replace(true); if !served { serving.await?; @@ -469,10 +432,64 @@ 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>>, +) -> Result { + 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)], timeout: Duration) { +async fn drain_actors(namespaces: &[&(String, Rc)], timeout: Duration) { let drains = namespaces.iter().map(|(class, namespace)| async move { let started = std::time::Instant::now(); let Drained { @@ -482,8 +499,10 @@ async fn drain_actors(namespaces: &[(String, Rc)], timeout: Dura 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, @@ -494,6 +513,7 @@ async fn drain_actors(namespaces: &[(String, Rc)], timeout: Dura ); } else { tracing::info!( + service, namespace = class.as_str(), drained, skipped, diff --git a/src/workerd/server/workerd.capnp b/src/workerd/server/workerd.capnp index ce408783050..aeb6b23803d 100644 --- a/src/workerd/server/workerd.capnp +++ b/src/workerd/server/workerd.capnp @@ -161,10 +161,11 @@ struct ClusterConfig { # 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 whose class defines one, all at once, 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, as it does without a `drain()`. 0 skips the `drain()` calls. Keep it well within the - # time the instance has to exit after SIGTERM, and under the lease (20 s). + # 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 { @@ -801,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 From 92651079e7e364a20e4345a7f793d9feb131d12f Mon Sep 17 00:00:00 2001 From: Gogo Date: Tue, 10 Nov 2026 08:06:35 +0000 Subject: [PATCH 6/7] Cluster test: drain() opt-in, a class without drain(), a throwing drain(), drainTimeoutMs = 0 Signed-off-by: Gogo Co-Authored-By: Claude Opus 5.5 --- .../server/tests/cluster/cluster-test.mjs | 159 +++++++++++++++++- .../server/tests/cluster/cluster-test.wd-test | 8 +- .../server/tests/cluster/cluster-worker.js | 32 +++- 3 files changed, 189 insertions(+), 10 deletions(-) diff --git a/src/workerd/server/tests/cluster/cluster-test.mjs b/src/workerd/server/tests/cluster/cluster-test.mjs index 80be4f62e73..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}`, @@ -693,8 +718,11 @@ test( // 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.*drained = 1/); - assert.match(owner.stderr, /cluster drain.*timed_out = 0/); + 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); @@ -726,7 +754,7 @@ test( elapsed >= DRAIN_TIMEOUT_MS - 100 && elapsed < DRAIN_TIMEOUT_MS + 5000, `exited ${elapsed} ms after SIGTERM` ); - assert.match(owner.stderr, /cluster drain.*timed_out = 1/); + 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. @@ -738,6 +766,123 @@ test( } ); +// 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 4eba21565b1..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" ), ) diff --git a/src/workerd/server/tests/cluster/cluster-worker.js b/src/workerd/server/tests/cluster/cluster-worker.js index f829c8b8241..35db238b3d4 100644 --- a/src/workerd/server/tests/cluster/cluster-worker.js +++ b/src/workerd/server/tests/cluster/cluster-worker.js @@ -74,6 +74,9 @@ // 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 @@ -97,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) { @@ -202,6 +209,7 @@ export class Counter extends DurableObject { // 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)) { @@ -329,6 +337,9 @@ export class Counter extends DurableObject { } 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 }); @@ -386,6 +397,9 @@ export class Counter extends DurableObject { 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. @@ -401,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 ?? '', + id: this.ctx.id.toString(), + }); + } +} From f32e62a04542c45d10996c47b57d88105f2c9d68 Mon Sep 17 00:00:00 2001 From: Gogo Date: Tue, 10 Nov 2026 08:20:17 +0000 Subject: [PATCH 7/7] cluster: a claim that a drain overtook does not keep its object Cluster::acquire_or_route checked may_keep before its awaits (advance, open_fresh, retries): a resolve could pass it, start_draining and abort_all could then run, and the resolve would insert a container that nothing aborts. It now checks again right before returning Route::Owned, after which the server inserts the container with no await between, and returns the draining error if the drain began. actor_drain_method also finds drain() without running the app's code: it walks the prototype chain with GetOwnPropertyDescriptor and counts only a data property that is a function; an accessor or a Proxy leaves it to the request. Signed-off-by: Gogo Co-Authored-By: Claude Opus 5.5 --- src/workerd/server/cluster/cluster.rs | 81 ++++++++++++++++++- src/workerd/server/cluster/ownership.rs | 2 +- .../server/factory/worker-factory-actor.c++ | 29 +++++-- 3 files changed, 105 insertions(+), 7 deletions(-) diff --git a/src/workerd/server/cluster/cluster.rs b/src/workerd/server/cluster/cluster.rs index 3ba399684b2..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) } @@ -459,6 +464,8 @@ 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; @@ -472,6 +479,51 @@ mod tests { 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, + hook: RefCell>>, + } + + #[async_trait::async_trait(?Send)] + impl crate::Store for Rc { + async fn get( + &self, + key: &str, + to: Option<&std::path::Path>, + ) -> crate::Result, 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, + metadata: &[(&str, String)], + ) -> crate::Result> { + self.inner.put(key, body, guard, metadata).await + } + + async fn list(&self, prefix: &str) -> crate::Result> { + 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 { let setup = Setup { url: String::new(), @@ -517,6 +569,33 @@ mod tests { }); } + #[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 3b6731fb06e..ed3a849c3c4 100644 --- a/src/workerd/server/cluster/ownership.rs +++ b/src/workerd/server/cluster/ownership.rs @@ -108,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")) } diff --git a/src/workerd/server/factory/worker-factory-actor.c++ b/src/workerd/server/factory/worker-factory-actor.c++ index 99efd7ff897..e7ebf53c2da 100644 --- a/src/workerd/server/factory/worker-factory-actor.c++ +++ b/src/workerd/server/factory/worker-factory-actor.c++ @@ -859,12 +859,31 @@ kj::Promise actor_drain_method(const ActorHandle& handle) { // The constructed instance; outside a request, anything else throws. KJ_IF_SOME(handler, lock.getExportedHandler(kj::none, kj::none, Frankenvalue(), actor, false)) { - auto self = jsg::JsObject(handler->self.getHandle(js)); - // A getter or a Proxy trap that throws leaves it to the request. + // 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([&] { - return self.get(js, "drain"_kj).tryCast() == kj::none - ? DrainMethod::None - : DrainMethod::Some; + auto context = js.v8Context(); + v8::Local name = js.strIntern("drain"_kj); + v8::Local get = js.strIntern("get"_kj); + v8::Local set = js.strIntern("set"_kj); + v8::Local value = js.strIntern("value"_kj); + v8::Local current = handler->self.getHandle(js); + for (auto depth = 0; depth < 64 && current->IsObject(); ++depth) { + auto object = current.As(); + if (object->IsProxy()) return DrainMethod::Unknown; + auto descriptor = jsg::check(object->GetOwnPropertyDescriptor(context, name)); + if (descriptor->IsObject()) { + auto fields = descriptor.As(); + 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;