Durable Object alarms in cluster mode (one host, several processes)

.github/workflows/botiverse-build.yml

@@ -0,0 +1,56 @@
+# Botiverse fork only: build the Linux x64 workerd binary for the cluster-alarms branch on a standard
+# GitHub runner. Upstream's builds use Cloudflare's 16-core runners and remote cache, which this fork
+# has neither of, so this is a plain from-scratch build (a few hours; Bazel's disk cache is kept
+# between runs to make later builds faster).
+name: Botiverse build
+
+on:
+  push:
+    branches:
+      - botiverse/**
+  workflow_dispatch:
+
+permissions:
+  contents: read
+
+concurrency:
+  group: botiverse-build-${{ github.ref }}
+  cancel-in-progress: true
+
+jobs:
+  linux-x64:
+    runs-on: ubuntu-24.04
+    # GitHub-hosted jobs stop at 6 hours.
+    timeout-minutes: 355
+    steps:
+      - uses: actions/checkout@v6
+        with:
+          persist-credentials: false
+      - name: Free disk space
+        run: |
+          sudo rm -rf /usr/share/dotnet /usr/local/lib/android /opt/ghc /opt/hostedtoolcache/CodeQL
+          df -h /
+      - name: Install the toolchain (clang, libc++ and lld 19+, tcl; see README.md)
+        run: |
+          sudo apt-get update -q
+          sudo apt-get install -y -q clang-19 libc++-19-dev libc++abi-19-dev lld-19 tcl8.6
+          # Bazel's sandbox only sees /bin, /usr/bin and /usr/local/bin.
+          sudo ln -sf /usr/bin/clang-19 /usr/local/bin/clang
+          sudo ln -sf /usr/bin/clang++-19 /usr/local/bin/clang++
+          sudo ln -sf /usr/bin/tclsh8.6 /usr/local/bin/tclsh
+      - uses: actions/cache@v5
+        with:
+          path: ~/bazel-disk-cache
+          key: botiverse-bazel-${{ runner.os }}-${{ runner.arch }}-${{ hashFiles('.bazelversion', '.bazelrc', 'MODULE.bazel') }}-${{ github.sha }}
+          restore-keys: |
+            botiverse-bazel-${{ runner.os }}-${{ runner.arch }}-${{ hashFiles('.bazelversion', '.bazelrc', 'MODULE.bazel') }}-
+      - name: Build
+        # The Android SDK was removed above (disk space); without these the Android rules find it empty and fail.
+        run: env -u ANDROID_HOME -u ANDROID_SDK_ROOT -u ANDROID_NDK_HOME -u ANDROID_NDK_ROOT -u ANDROID_NDK_LATEST_HOME bazel build //src/workerd/server:workerd --config=release --disk_cache=~/bazel-disk-cache --jobs=4
+      - name: Check the binary runs
+        run: bazel-bin/src/workerd/server/workerd --version
+      - uses: actions/upload-artifact@v7
+        with:
+          name: workerd-linux-x64-${{ github.sha }}
+          path: bazel-bin/src/workerd/server/workerd
+          retention-days: 30

compile_flags.txt

@@ -21,7 +21,10 @@
 -isystembazel-bin/external/+http+capnp-cpp/src/capnp/compat/_virtual_includes/http-over-capnp
 -isystembazel-bin/external/+http+capnp-cpp/src/capnp/compat/_virtual_includes/json
 -isystembazel-bin/external/+http+capnp-cpp/src/kj/_virtual_includes/kj
--isystembazel-bin/external/+http+capnp-cpp/src/kj/_virtual_includes/kj-async
+-isystembazel-bin/external/+http+capnp-cpp/src/kj/_virtual_includes/kj-async-core
+-isystembazel-bin/external/+http+capnp-cpp/src/kj/_virtual_includes/kj-async-io
+-isystembazel-bin/external/+http+capnp-cpp/src/kj/_virtual_includes/kj-async-os-hdrs
+-isystembazel-bin/external/+http+capnp-cpp/src/kj/_virtual_includes/kj-test
 -isystembazel-bin/external/+http+capnp-cpp/src/kj/compat/_virtual_includes/kj-brotli
 -isystembazel-bin/external/+http+capnp-cpp/src/kj/compat/_virtual_includes/kj-gzip
 -isystembazel-bin/external/+http+capnp-cpp/src/kj/compat/_virtual_includes/kj-http

src/workerd/api/restore.c++

@@ -495,46 +495,78 @@ kj::Promise<WorkerInterface::CustomEvent::Result> RestoreRpcStubCustomEvent::run
   }
 }
 
-kj::Promise<WorkerInterface::CustomEvent::Result> RestoreRpcStubCustomEvent::sendRpc(
-    capnp::HttpOverCapnpFactory& httpOverCapnpFactory,
-    capnp::ByteStreamFactory& byteStreamFactory,
-    FrankenvalueHandler& frankenvalueHandler,
-    rpc::EventDispatcher::Client dispatcher) {
-  // This contains a lot of similar code to JsRpcSessionCustomEvent::sendRpc() for setting
-  // up membranes, handling the returned JsRpcSession, etc.
+namespace {
 
-  auto revokePaf = kj::newPromiseAndFulfiller<void>();
+kj::Promise<void> remoteRpcSessionTask(rpc::JsRpcSession::Client session,
+    kj::Promise<void> allStubsDropped,
+    kj::Own<kj::PromiseFulfiller<void>> revoker) {
+  // Revoke all stubs in the session as soon as this task completes or is canceled. Normally the
+  // server side doesn't end the session while any stubs still exist, so this only makes a
+  // difference when some sort of error occurred. The stubs are probably already broken in that
+  // case, but revoking them ensures the underlying transport isn't held open waiting for the JS
+  // garbage collector to actually collect the JsRpcStub objects.
   KJ_DEFER({
-    if (revokePaf.fulfiller->isWaiting()) {
-      revokePaf.fulfiller->reject(KJ_EXCEPTION(DISCONNECTED, "JS-RPC-restore session canceled"));
+    if (revoker->isWaiting()) {
+      revoker->reject(KJ_EXCEPTION(DISCONNECTED, "JS-RPC session canceled"));
     }
   });
 
-  auto req = dispatcher.restoreRpcStubRequest();
-  frankenvalueHandler.toCapnp(restoreParams, req.initParams());
-  auto sent = req.send();
+  try {
+    // Wait for `session` to resolve to a null capability, which the server does once the session
+    // is over. (If the server is using the "old approach" of not returning a `session` at all, the
+    // return itself represents the end of the session and `session` resolves to null just the
+    // same.)
+    co_await session.whenResolved().exclusiveJoin(kj::mv(allStubsDropped));
+  } catch (...) {
+    auto e = kj::getCaughtExceptionAsKj();
+    if (revoker->isWaiting()) {
+      revoker->reject(e.clone());
+    }
+    kj::throwFatalException(kj::mv(e));
+  }
+}
 
-  rpc::JsRpcTarget::Client cap = sent.getTarget();
+}  // namespace
 
+IoChannelFactory::RpcChannel::Session wrapRemoteRpcSession(
+    rpc::JsRpcTarget::Client cap, rpc::JsRpcSession::Client session) {
+  auto revokePaf = kj::newPromiseAndFulfiller<void>();
   cap = capnp::membrane(kj::mv(cap), kj::refcounted<RevokerMembrane>(kj::mv(revokePaf.promise)));
 
+  // When no more stubs exist in the session, proactively end it. This matters in particular when
+  // the client drops the stub without making any calls at all, e.g. because serializing the
+  // arguments failed: `cap` may be a pipelined capability on a call that has not returned yet, in
+  // which case simply dropping it is not detected by the server.
   auto completionPaf = kj::newPromiseAndFulfiller<void>();
   cap = capnp::membrane(
       kj::mv(cap), kj::refcounted<CompletionMembrane>(kj::mv(completionPaf.fulfiller)));
 
-  capFulfiller->fulfill(kj::mv(cap));
+  return {
+    .cap = kj::mv(cap),
+    .task = remoteRpcSessionTask(
+        kj::mv(session), kj::mv(completionPaf.promise), kj::mv(revokePaf.fulfiller)),
+  };
+}
+
+kj::Promise<WorkerInterface::CustomEvent::Result> RestoreRpcStubCustomEvent::sendRpc(
+    capnp::HttpOverCapnpFactory& httpOverCapnpFactory,
+    capnp::ByteStreamFactory& byteStreamFactory,
+    FrankenvalueHandler& frankenvalueHandler,
+    rpc::EventDispatcher::Client dispatcher) {
+  auto req = dispatcher.restoreRpcStubRequest();
+  frankenvalueHandler.toCapnp(restoreParams, req.initParams());
+  auto sent = req.send();
+
+  auto session = wrapRemoteRpcSession(sent.getTarget(), sent.getSession());
 
-  auto session = sent.getSession();
+  // `session.task` already propagates any errors from the call, so the call promise itself can be
+  // dropped. (It would NOT work to use `req.sendForPipeline()` above, since pipelined capabilities
+  // cannot resolve until the call returns, but `sendForPipeline()` explicitly inhibits the return
+  // message.)
   { auto drop = kj::mv(sent); }
-  try {
-    co_await session.whenResolved().exclusiveJoin(kj::mv(completionPaf.promise));
-  } catch (...) {
-    auto e = kj::getCaughtExceptionAsKj();
-    if (revokePaf.fulfiller->isWaiting()) {
-      revokePaf.fulfiller->reject(e.clone());
-    }
-    kj::throwFatalException(kj::mv(e));
-  }
+
+  capFulfiller->fulfill(kj::mv(session.cap));
+  co_await session.task;
 
   co_return WorkerInterface::CustomEvent::Result{.outcome = EventOutcome::OK};
 }

src/workerd/api/restore.h

@@ -55,6 +55,17 @@ jsg::Promise<jsg::Value> restoreCurrentEntrypoint(jsg::Lock& js,
     const jsg::TypeHandler<jsg::Ref<Fetcher>>& fetcherHandler,
     const jsg::TypeHandler<jsg::Ref<JsRpcStub>>& rpcStubHandler);
 
+// Wraps the `target` and `session` capabilities returned by an RPC that opened a JS RPC session on
+// a remote worker (`EventDispatcher.restoreRpcStub()`, `WorkerdDebugPort.getRpcTargetFromToken()`)
+// into an `RpcChannel::Session`. The returned `cap` is membraned so that dropping every stub
+// derived from it ends the session promptly, without waiting for the server to notice, and so that
+// completing or cancelling the returned `task` revokes any stubs still outstanding, so they cannot
+// hold the underlying transport open. `task` completes when `session` resolves (to null, i.e. the
+// server ended the session) or all local stubs are dropped, whichever comes first, and fails if the
+// session fails.
+IoChannelFactory::RpcChannel::Session wrapRemoteRpcSession(
+    rpc::JsRpcTarget::Client target, rpc::JsRpcSession::Client session);
+
 // Custom event that calls the `[restore]()` method of the entrypoint and expects that the result
 // is a `Fetcher`. (When the result is expected to be an RpcStub instead, we use
 // `RestoreRpcStubCustomEvent`.)

src/workerd/io/actor-sqlite-test.c++

@@ -2569,42 +2569,36 @@ KJ_TEST("unconfirmed setAlarm failure still breaks output gate") {
   KJ_EXPECT_THROW_MESSAGE("alarm commit failed", promise.wait(test.ws));
 }
 
+// Drives `test` into the broken state through SqliteDatabase's critical-error path: a write that
+// fails with SQLITE_NOMEM inside `txn` makes SQLite auto-rollback the transaction, which invokes
+// ActorSqlite::onCriticalError(). The global heap limit is restored before returning so that
+// connections opened later are unaffected.
+void breakViaCriticalError(ActorSqliteTest& test, ActorCacheInterface::Transaction& txn) {
+  auto heapLimit = test.db.run("PRAGMA hard_heap_limit").getInt64(0);
+  KJ_DEFER(sqlite3_hard_heap_limit64(heapLimit));
+
+  test.db.run("PRAGMA hard_heap_limit=8192");  // 8KB limit
+
+  // A value large enough to exceed the limit. Copied once more to convert Array<byte> to
+  // Array<const byte>.
+  auto largeData = kj::heapArray<byte>(50000, 'X');
+  try {
+    txn.put(kj::str("large_key"), kj::heapArray<const byte>(largeData.asBytes()), {}, nullptr);
+    KJ_FAIL_ASSERT("Query should have failed with SQLITE_NOMEM");
+  } catch (kj::Exception& e) {
+    KJ_ASSERT(e.getDescription().contains("SENTRY_DO"), e);
+    KJ_ASSERT(e.getDescription().contains("out of memory"), e);
+  }
+}
+
 KJ_TEST("sync() throws after critical error in explicit transaction") {
   ActorSqliteTest test({.monitorOutputGate = false});
-  auto heapLimit = [&]() {
-    auto row = test.db.run("PRAGMA hard_heap_limit");
-    return row.getInt(0);
-  }();
-  KJ_DBG(heapLimit);
-  KJ_DEFER(sqlite3_hard_heap_limit64(heapLimit););
 
-  // Start an explicit transaction
+  // Start an explicit transaction and do a write within it.
   auto txn = test.startTransaction();
-
-  // Do a write within the transaction
   txn->put(kj::str("foo"), kj::heapArray("bar"_kj.asBytes()), {}, nullptr);
 
-  // Trigger a critical error using SQLITE_NOMEM by setting a very low heap limit
-  // and then trying to insert a large value.
-  try {
-    // Set SQLite's memory limit very low to trigger SQLITE_NOMEM
-    test.db.run("PRAGMA hard_heap_limit=8192");  // 8KB limit
-
-    // Create data that will exceed the memory limit
-    auto largeData = kj::heapArray<byte>(50000, 'X');  // 50KB
-
-    // This should trigger SQLITE_NOMEM, causing SQLite to auto-rollback the transaction,
-    // which will trigger the critical error handler.
-    //
-    // We have to copy it again in order to convert to a Array<const byte> from an Array<byte>.
-    txn->put(kj::str("large_key"), kj::heapArray<const byte>(largeData.asBytes()), /*options=*/{},
-        nullptr);
-    KJ_FAIL_ASSERT("Query should have failed with SQLITE_NOMEM");
-  } catch (kj::Exception& e) {
-    // Expected: out of memory error. We catch and ignore this to continue the test.
-    KJ_ASSERT(e.getDescription().contains("SENTRY_DO"));
-    KJ_ASSERT(e.getDescription().contains("out of memory"));
-  }
+  breakViaCriticalError(test, *txn);
 
   // sync() should also throw an exception because the storage is now broken
   auto syncResult = test.sync();
@@ -2617,6 +2611,112 @@ KJ_TEST("sync() throws after critical error in explicit transaction") {
   KJ_EXPECT_THROW_MESSAGE("broken", txn->commit());
 }
 
+// Asserts that `test.db` has been closed. Direct use fails, and a second connection to the same
+// file can take an exclusive lock (which the closed connection's open transaction would have
+// blocked) and sees the committed write but not the uncommitted one.
+void expectDatabaseClosed(ActorSqliteTest& test) {
+  KJ_EXPECT_THROW_MESSAGE("database has been closed", test.db.run("SELECT 1"));
+
+  SqliteDatabase fresh(test.vfs, kj::Path({"foo"}), kj::WriteMode::MODIFY);
+  fresh.run("BEGIN EXCLUSIVE");
+  KJ_EXPECT(fresh.run("SELECT count(*) FROM _cf_KV WHERE key = 'committed'").getInt(0) == 1);
+  KJ_EXPECT(fresh.run("SELECT count(*) FROM _cf_KV WHERE key = 'uncommitted'").getInt(0) == 0);
+  fresh.run("COMMIT");
+}
+
+KJ_TEST("shutdown() leaves the connection open; the owner closes it afterwards") {
+  ActorSqliteTest test({.monitorOutputGate = false});
+
+  test.put("committed", "yes");
+  test.pollAndExpectCalls({"commit"})[0]->fulfill();
+
+  // Leave a second write open in an implicit transaction whose commit has not run yet.
+  test.put("uncommitted", "yes");
+
+  test.actor.shutdown(kj::none);
+
+  // Storage is broken, but shutdown() must not close the connection itself: embedders may call it
+  // from inside a SQLite callback, where closing is unsafe. The owner closes it later.
+  KJ_EXPECT_THROW_MESSAGE(ActorCache::SHUTDOWN_ERROR_MESSAGE, test.actor.getSqliteKv());
+  test.db.run("SELECT 1");
+
+  // The scheduled commit observes the shutdown and breaks the output gate with the shutdown
+  // exception rather than committing.
+  KJ_EXPECT_THROW_MESSAGE(ActorCache::SHUTDOWN_ERROR_MESSAGE, test.gate.onBroken().wait(test.ws));
+  test.pollAndExpectCalls({});
+
+  // Closing with the implicit transaction still open rolls it back and releases the file.
+  test.db.close();
+  expectDatabaseClosed(test);
+
+  // A second shutdown() on a closed database is a no-op.
+  test.actor.shutdown(kj::none);
+  KJ_EXPECT_THROW_MESSAGE(ActorCache::SHUTDOWN_ERROR_MESSAGE, test.actor.getSqliteKv());
+  expectDatabaseClosed(test);
+}
+
+KJ_TEST("shutdown() after a critical error keeps the original exception") {
+  ActorSqliteTest test({.monitorOutputGate = false});
+
+  test.put("committed", "yes");
+  test.pollAndExpectCalls({"commit"})[0]->fulfill();
+
+  auto txn = test.startTransaction();
+  txn->put(kj::str("uncommitted"), kj::heapArray("yes"_kj.asBytes()), {}, nullptr);
+  breakViaCriticalError(test, *txn);
+  KJ_EXPECT_THROW_MESSAGE("broken.outputGateBroken", test.actor.getSqliteKv());
+
+  auto expectOriginalBrokenException = [&]() {
+    KJ_IF_SOME(e, kj::runCatchingExceptions([&]() { test.actor.getSqliteKv(); })) {
+      // The first exception wins: the shutdown exception must not replace the storage error.
+      KJ_EXPECT(e.getDescription().contains("broken.outputGateBroken"), e);
+      KJ_EXPECT(!e.getDescription().contains(ActorCache::SHUTDOWN_ERROR_MESSAGE), e);
+    } else {
+      KJ_FAIL_EXPECT("getSqliteKv() should have thrown");
+    }
+  };
+
+  // shutdown() runs after the break (the broken output gate aborts the IoContext), then the owner
+  // closes the connection.
+  test.actor.shutdown(kj::none);
+  expectOriginalBrokenException();
+  test.db.close();
+  expectDatabaseClosed(test);
+
+  test.actor.shutdown(kj::none);
+  expectOriginalBrokenException();
+}
+
+KJ_TEST("shutdown() during in-flight commit does not unschedule the persisted alarm") {
+  ActorSqliteTest test({.monitorOutputGate = false});
+
+  // Persist an alarm at 2ms, so that the next write moves the alarm earlier and thus starts a
+  // precommit scheduling request that the commit must wait on.
+  test.setAlarm(twoMs);
+  test.pollAndExpectCalls({"scheduleRun(2ms)"})[0]->fulfill();
+  test.pollAndExpectCalls({"commit"})[0]->fulfill();
+
+  // Move the alarm earlier. The local transaction commits, then commitImpl() suspends waiting for
+  // the scheduler to confirm the 1ms request.
+  test.setAlarm(oneMs);
+  auto scheduleFulfiller = kj::mv(test.pollAndExpectCalls({"scheduleRun(1ms)"})[0]);
+
+  // Shut down while that commit is parked.
+  test.actor.shutdown(kj::none);
+
+  // Resume the commit. Now that storage is broken, it must not consult the database's alarm state
+  // and must not issue any further scheduler requests. In particular it must not issue
+  // `scheduleRun(none)`, which would cancel the alarm that the database still persists.
+  scheduleFulfiller->fulfill();
+  test.pollAndExpectCalls({});
+  KJ_EXPECT_THROW_MESSAGE(ActorCache::SHUTDOWN_ERROR_MESSAGE, test.gate.onBroken().wait(test.ws));
+
+  // The database still holds the 1ms alarm for the next instance to pick up.
+  SqliteDatabase fresh(test.vfs, kj::Path({"foo"}), kj::WriteMode::MODIFY);
+  KJ_EXPECT(fresh.run("SELECT value FROM _cf_METADATA WHERE key = 1").getInt64(0) ==
+      (oneMs - kj::UNIX_EPOCH) / kj::NANOSECONDS);
+}
+
 KJ_TEST("allowUnconfirmed put in explicit transaction does not block output gate") {
   ActorSqliteTest test;