Bidirectional calling with Durable Objects (Error: RPC stub used after being disposed)

[bndkt]

First of all, this is an awesome library and I'm heavily experimenting with very interesting use cases.

One thing that I still struggle with after lots of experimentation is bidirectional calling and I'd like to clarify if what I'm aiming for is even intended to work. Simply put, I'd like to register a callback from a client in a Durable Object, which then executes the callback(s) upon certain events happening.

Here are some relevant sections of the code:

Server (Durable Object)

import { DurableObject, RpcStub } from "cloudflare:workers";

export class MyRpcServer extends DurableObject {
  #subscriptions: Set<RpcStub<(message: string) => void>> = new Set();

  async addMessage(message: string) {
    console.log("DO: Adding message:", message);
    console.log("DO: Number of subscribers:", this.#subscriptions.size);

    for (const sub of this.#subscriptions) {
      console.log("DO: Sending message to subscriber");
      await sub(message);
      console.log("DO: Should be sent");
    }
  }

  subscribe(callback: RpcStub<(message: string) => void>) {
    console.log("DO: Subscribing");

    this.#subscriptions.add(callback);

    return () => {
      console.log("DO: Unsubscribing");
      this.#subscriptions.delete(callback);
    };
  }
}

Client

const localReceiveMessage = (message: string) => {
  console.log("Adding message", message);
};

let api = newWebSocketRpcSession<MyRpcServer>("ws://localhost:5173/rpc");
api.subscribe(new RpcStub<(message: string) => void>(localReceiveMessage))

If I execute the callback immediately within the subscribe function, it works. But if I store it and then execute it later (upon receiving a message via addMessage within the DO), it fails with Error: RPC stub used after being disposed.. I have read the README section about automatic disposal and also tried lots of variations of the code, duplicating the stubs with dup(), but to no avail so far.

I've created a reproducible example here: https://github.com/bndkt/rpc-example and would appreciate any pointes on whether this should be a supported scenario at all and if yes, where I'm missing something.

[kentonv]

Stubs received in the parameters of an RPC call are automatically disposed when the call returns. If you want to keep the stub around beyond that, you need to .dup() it and store the duplicate. So:

-    this.#subscriptions.add(callback);
+    this.#subscriptions.add(callback.dup());

This is covered in the readme but still seems to be a common point of confusion. Perhaps we should improve the error message to make it more clear what you need to do...

[bndkt]

Thanks for the quick reply. I'm pretty sure I've tried .dup() everywhere, but let me verify and get back on that.

[bndkt]

I added the duplication (https://github.com/bndkt/rpc-example/blob/c145e6fe9f2d1ed91b9600c2f6b0ef562a039e7c/src/MyRpcServer.ts#L26) but still get the error. Interestingly, it still errors but the error has changed to Error: The execution context which hosts this callback is no longer running. I've simplified the client code to rule out any React shenanigans (https://github.com/bndkt/rpc-example/blob/6b91ff9c338b9781c410c937968a25adc578574a/src/components/Messages.tsx#L12). Do I have to .dup() on the client side as well and maybe assign to a variable? Have tried that as well but maybe in a wrong way.

[bndkt]

@kentonv I'd like to take a step back and verify whether the setting up and passing of the stubs is done how you intended it to be used. From a typing perspective, here's what makes sense for me. Would be great if you could confirm whether this make sense:

On the client side I have a sync function than when I pass it to the server, I have a sync function param there and therefor would not use await on the server side which then means nothing happens. So I types the callback as an RpcStub and on the client turn the local sync function into an RpcStub first:

const api = newWebSocketRpcSession<MyRpcServer>("/rpc");

const addMessage = (message: string): void => {
  console.log("Adding message", message);
};

const cb = new RpcStub<(message: string) => void>(addMessage);
api.subscribe(cb);

Now on the server side I store a dup of the RpcStub I received:

type Callback = RpcStub<(message: string) => void>;

export class MyRpcServer extends DurableObject {
  #subscriptions: Set<Callback> = new Set();

async addMessage(message: string) {
    for (const sub of this.#subscriptions) {
      await sub(message);
    }
  }

  async subscribe(callback: Callback) {
    console.log("DO: Subscribing");

    const cb = new RpcStub(callback);

    await cb("immediate test upon subscription");

    this.#subscriptions.add(cb.dup());

    return async () => {
      console.log("DO: Unsubscribing");
      this.#subscriptions.delete(callback);
    };
  }
}

Does this make sense?

The message that I sent immediately upon subscription (via await cb()) comes through on the client, but any further messages sent via "addMessage" result in the Error: RPC stub used after being disposed.message being displayed on the client side (in the browser!).

[kentonv]

const cb = new RpcStub(callback);

You don't need new RpcStub here: callback is already an RpcStub. This is probably causing problems: you are creating a second RpcStub that wraps the original stub. Then you call .dup(), but only on the outer wrapper. This creates a duplicate of the wrapper, but not of the thing that it is wrapping. The inner callback still gets disposed when the method returns, and thus you still get the "stub used after being disposed" error.

(BTW, you don't need new RpcStub on the client side either. The system automatically wraps functions in RpcStub when passing across an RPC boundary. But in that case it's not causing a problem.)

Error: The execution context which hosts this callback is no longer running.

This error is actually produced by the Workers Runtime. It indicates that the proxy worker between the browser and the Durable Object has shut down, severing the connection to the callback. You haven't shared any code from that proxy worker, but I suspect the problem is there. Can you share it?

[bndkt]

Thanks, I'll try this once I get back to my desk later.

The worker code is in the reproducible example I shared: https://github.com/bndkt/rpc-example

But I'll try your suggestions and report back.

[kentonv]

https://github.com/bndkt/rpc-example

This link is 404 for me, do you need to make it public?

[bndkt]

I tried this (and also updated the code here: https://github.com/bndkt/rpc-example), but still get the same errors.

Here's the worker code for reference:

import { newWorkersRpcResponse } from "capnweb";

export { MyRpcServer } from "./MyRpcServer";

export default {
  async fetch(request: Request, env: Env, _ctx: ExecutionContext) {
    const url = new URL(request.url);

    if (url.pathname.startsWith("/rpc")) {
      const durable = env.MyRpcServer.getByName("test");
      return newWorkersRpcResponse(request, durable);
    }

    return env.ASSETS.fetch(url);
  },
};

DO is now:

import { DurableObject } from "cloudflare:workers";

type Callback = (message: string) => void;

export class MyRpcServer extends DurableObject {
  #subscriptions: Set<Callback> = new Set();

  async addMessage(message: string) {
    console.log("DO: Adding message:", message);

    for (const sub of this.#subscriptions) {
      console.log("DO: Sending message to subscriber");
      await sub(message);
      console.log("DO: Should be sent");
    }
  }

  async subscribe(cb: Callback) {
    console.log("DO: Subscribing");

    await cb("immediate test upon subscription");

    this.#subscriptions.add(cb.dup());

    return async () => {
      console.log("DO: Unsubscribing");
      this.#subscriptions.delete(cb);
    };
  }
}

Btw. the reason that I went to create an RpcStub on the client in the first place is that I now get type errors because for this.#subscriptions.add(cb.dup());, cb has not dup() function and also TS now thinks cb is a sync function (which it is on the client).

Here's the client code for completeness:

import type { MyRpcServer } from "../MyRpcServer";

const api = newWebSocketRpcSession<MyRpcServer>("/rpc");

const addMessage = (message: string): void => {
  console.log("Adding message", message);
};

api.subscribe(addMessage)

[bndkt]

https://github.com/bndkt/rpc-example

This link is 404 for me, do you need to make it public?

🤦‍♂️ My bad, it's public now.

[kentonv]

Btw. the reason that I went to create an RpcStub on the client in the first place is that I now get type errors because for this.#subscriptions.add(cb.dup());, cb has not dup() function and also TS now thinks cb is a sync function (which it is on the client).

You should still declare the function like:

  async subscribe(cb: RpcStub<Callback>) {

But on the client side, you should be able to pass in the type Callback without explicitly wrapping it in RpcStub. This is because the typescript type RpcStub<MyRpcServer> automatically transforms method arguments and return types to add and remove RpcStub<>.

At least, that's how it's supposed to work, but the TypeScript types do have some issues that still need to be worked out.

Error: The execution context which hosts this callback is no longer running.

If you're still seeing this error, it suggests that the client has already disconnected from the Worker before the callback was invoked. When the client disconnects, the stateless Worker (fetch event handler) stops (the "execution context" terminates, hence the error). Of course, if the client is disconnected, you can't invoke the callback anymore.

[bndkt]

I don't think the connection being disconnected is the problem as I can see the WS connections is still alive in Chrome Dev Tools and when I do a simple RPC (the other button I have in their to do a one way call from client to server to return a string) it uses the existing WS connection.

Also interestingly I sometimes get Error: The execution context which hosts this callback is no longer running., other times I get Error: Attempted to use RPC stub after it has been disposed. (without changing anything).

Not to dwell on the types but I think they might be a clue (at least as long as the types are erroring I keep thinking "I'm holding this thing wrong").

I get the following type error when passing the callback function to the subscribe function.

When turning the callback function ("addMessage") into an async function, I get another error in the client all together btw:

const addMessage = async (message: string): Promise<void> => {
  console.log("Adding message", message);
};

[bndkt]

Just checking, is my base assumption correct that I can just use a Durable Object as an RpcTarget?

import { newWorkersRpcResponse } from "capnweb";

export { MyRpcServer } from "./MyRpcServer";

export default {
  async fetch(request: Request, env: Env, _ctx: ExecutionContext) {
    const url = new URL(request.url);

    if (url.pathname.startsWith("/rpc")) {
      const durable = env.MyRpcServer.getByName("test");
      return newWorkersRpcResponse(request, durable);
    }

    return env.ASSETS.fetch(url);
  },
};

[myuzumaki]

Just checking, is my base assumption correct that I can just use a Durable Object as an RpcTarget?

I can't answer that question, but here's how I'm accessing my Durable Object via an RpcTarget. It works, but I don't know if it's a good idea or not to do it this way.

1. The DurableObject creates the instance of the RpcTarget and passes a reference to itself (this).
2. The RpcTarget instance calls functions on the DurableObject via the reference passed in the constructor.
3. Might need to manually clean-up these instances, hence I added a terminate() function that the client can call.

[bndkt]

Thanks for providing this example @myuzumaki! One question, do you do any bidirectional calling where you store the stub on the server side (within the DO) and re-use it later, e.g. upon events happening?

[myuzumaki]

One question, do you do any bidirectional calling where you store the stub on the server side (within the DO) and re-use it later, e.g. upon events happening?

I tried this out, and I get the same result as you. If I call the callback or the duplicated callback immediately, it works. If I store the callback and try calling it later... I get this error:

client: capnweb.js?v=6027f409:1331` Uncaught (in promise) Error: Attempted to use RPC stub after it has been disposed.

DurableObject: Uncaught Error: Attempted to use RPC stub after it has been disposed. Error at null.
(file:/// blah blah /node_modules/.pnpm/capnweb@0.2.0/node_modules/capnweb/src/core.ts:252:5)

// works as expected both times
async registerCallback(clientCallback: RpcStub) {
  await clientCallback('your callback has been registered!');
  this.savedCallback = await clientCallback.dup();
  await this.savedCallback('testing the dup callback');
}

// fails 
async testCallback() {
    await this.savedCallback('Hello from the DO!');
}

EDIT: I tried this in both the DurableObject and in the RpcTarget created by the DO (called myRPC in my screenshot above)... same results both times.

// as soon as registerCallback() finishes we will get "stub RPC broken" message 
async registerCallback(clientCallback: RpcStub) {
    await clientCallback('callback registered by the RpcTarget');
    this.myCallback = clientCallback.dup();

    this.myCallback.onRpcBroken((error: any) => {
      console.warn('stub RPC broken');
      console.error(error);
    });
}

Edit 2: Adding the onRpcBroken() callback shows that as soon as my registerCallback() function finishes, the error (Attempted to use RPC stub after it has been disposed) is thrown - even though I have not yet attempted to use the RPC stub.