リモートプロシージャコール（RPC）

Workers には、JavaScript ネイティブの RPC（リモートプロシージャコール） ↗ システムが組み込まれており、次ができます。

• Worker に公開メソッドを定義し、同じ Cloudflare アカウント上のほかの Workers から Service Bindings 経由で呼び出せます
• Durable Objects に公開メソッドを定義し、それに対するバインディングを宣言した、同じ Cloudflare アカウント上のほかの Workers から呼び出せます

RPC システムは、同じ Worker 内の JavaScript 関数を呼ぶ感覚にできるだけ近づけて設計されています。ほとんどの場合、すべてが 1 つの Worker にあるときと同じ書き方でコードを書けます。

たとえば、Worker B が公開メソッド add(a, b) を実装している場合です。

{
	"$schema": "./node_modules/wrangler/config-schema.json",
	"name": "worker_b",
	"main": "./src/workerB.js"
}

"$schema" = "./node_modules/wrangler/config-schema.json"
name = "worker_b"
main = "./src/workerB.js"

import { WorkerEntrypoint } from "cloudflare:workers";

export default class extends WorkerEntrypoint {
	async fetch() {
		return new Response("Hello from Worker B");
	}

	add(a, b) {
		return a + b;
	}
}

import { WorkerEntrypoint } from "cloudflare:workers";

export default class extends WorkerEntrypoint {
	async fetch() {
		return new Response("Hello from Worker B");
	}

	add(a: number, b: number) {
		return a + b;
	}
}

from workers import WorkerEntrypoint, Response

class Default(WorkerEntrypoint):
    async def fetch(self, request):
        return Response("Hello from Worker B")

    def add(self, a: int, b: int) -> int:
        return a + b

Worker A は、Worker B への バインディング を宣言できます。

{
	"$schema": "./node_modules/wrangler/config-schema.json",
	"name": "worker_a",
	"main": "./src/workerA.js",
	"services": [
		{
			"binding": "WORKER_B",
			"service": "worker_b"
		}
	]
}

"$schema" = "./node_modules/wrangler/config-schema.json"
name = "worker_a"
main = "./src/workerA.js"

[[services]]
binding = "WORKER_B"
service = "worker_b"

これにより、Worker A から Worker B の add() メソッドを呼び出せます。

export default {
	async fetch(request, env) {
		const result = await env.WORKER_B.add(1, 2);
		return new Response(result);
	},
};

export default {
	async fetch(request, env) {
		const result = await env.WORKER_B.add(1, 2);
		return new Response(result);
	},
};

from workers import WorkerEntrypoint, Response

class Default(WorkerEntrypoint):
    async def fetch(self, request):
        result = await self.env.WORKER_B.add(1, 2)
        return Response(f"Result: {result}")

この場合のクライアント（Worker A）は Worker B を呼び、クライアントが渡した引数で特定の手続きを実行するよう指示します。これは標準の JavaScript クラスで実現します。

呼び出すメソッドがサーバー側で非同期宣言されていなくても、クライアント側では非同期として振る舞います。結果は await する必要があります。

RPC 呼び出しが実際に返すのは Promise ではありませんが、Promise のように振る舞う型です。この型は「カスタム thenable」で、then() メソッドを実装しています。JavaScript は任意の「thenable」型を await できるため、ほとんどの場合、戻り値を Promise のように扱えます。

（この型が実際の Promise でない理由は、後ほど説明します。）

Structured Clone 可能 ↗ な型のほぼすべてを、RPC メソッドのパラメーターまたは戻り値に使えます。オブジェクト、配列、文字列、数値など、JavaScript の基本的な「値」型のほとんどを含みます。

Structured Clone の例外として、アプリ定義のクラス（またはカスタムプロトタイプを持つオブジェクト）は、後述の場合を除き RPC で渡せません。

RPC システムは、Structured Clone 可能でない次の型もサポートします。

• 関数。送信元へコールバックするスタブに置き換えられます。
• RpcTarget を拡張するアプリ定義クラス。同様にスタブに置き換えられます。
• ReadableStream と WritableStream。ストリーミングのフロー制御は自動です。
• Request と Response。HTTP メッセージを扱いやすく表します。
• RPC スタブ自体。サードの Worker から受け取ったスタブでも構いません。

関数を RPC で送れます。送ると、関数は「スタブ」に置き換わります。受け手はスタブを関数のように呼べますが、呼ぶと関数の発生元へ新しい RPC が戻ります。

次の 2 つの Workers は、Service Binding でつながっています。カウンターサービスは RPC メソッド newCounter() を提供し、関数を返します。

{
	"$schema": "./node_modules/wrangler/config-schema.json",
	"name": "counter-service",
	"main": "./src/counterService.js"
}

"$schema" = "./node_modules/wrangler/config-schema.json"
name = "counter-service"
main = "./src/counterService.js"

import { WorkerEntrypoint } from "cloudflare:workers";

export default class extends WorkerEntrypoint {
	async fetch() {
		return new Response("Hello from counter-service");
	}

	async newCounter() {
		let value = 0;
		return (increment = 0) => {
			value += increment;
			return value;
		};
	}
}

import { WorkerEntrypoint } from "cloudflare:workers";

export default class extends WorkerEntrypoint {
	async fetch() {
		return new Response("Hello from counter-service");
	}

	async newCounter() {
		let value = 0;
		return (increment = 0) => {
			value += increment;
			return value;
		};
	}
}

from workers import WorkerEntrypoint, Response

class Default(WorkerEntrypoint):
    async def fetch(self, request):
        return Response("Hello from counter-service")

    async def new_counter(self):
        value = 0
        def increment(amount=0):
            nonlocal value
            value += amount
            return value
        return increment

クライアント Worker から、この関数を呼び出せます。

{
	"$schema": "./node_modules/wrangler/config-schema.json",
	"name": "client_worker",
	"main": "./src/clientWorker.js",
	"services": [
		{
			"binding": "COUNTER_SERVICE",
			"service": "counter-service"
		}
	]
}

"$schema" = "./node_modules/wrangler/config-schema.json"
name = "client_worker"
main = "./src/clientWorker.js"

[[services]]
binding = "COUNTER_SERVICE"
service = "counter-service"

export default {
	async fetch(request, env) {
		using f = await env.COUNTER_SERVICE.newCounter();
		await f(2); // returns 2
		await f(1); // returns 3
		const count = await f(-5); // returns -2

		return new Response(count);
	},
};

export default {
	async fetch(request: Request, env: Env) {
		using f = await env.COUNTER_SERVICE.newCounter();
		await f(2); // returns 2
		await f(1); // returns 3
		const count = await f(-5); // returns -2

		return new Response(count);
	},
};

from workers import WorkerEntrypoint, Response

class Default(WorkerEntrypoint):
    async def fetch(self, request):
        counter = await self.env.COUNTER_SERVICE.new_counter()
        await counter(2)  # returns 2
        await counter(1)  # returns 3
        count = await counter(-5)  # returns -2

        return Response(str(count))

なぜこのようなことができるのでしょうか。システムは関数そのものをシリアライズしていません。CounterService が返した関数が呼ばれると、別の Worker から呼ばれた場合でも、その関数は CounterService 内で実行されます。

内部では、呼び出し元は関数そのものを直接呼んでいるのではなく、「スタブ（stub）」と呼ばれるものを呼んでいます。スタブは Proxy ↗ オブジェクトであり、クライアントはリモートサービスを、同じ Worker 内のローカルな処理であるかのように呼び出せます。裏側では、CounterService を実装している Worker に戻り、先に返されていた関数クロージャーの実行を依頼します。

RPC のパラメーターとして関数を送ることもできます。「サーバー」が「クライアント」へコールバックでき、関係の向きが逆転します。

このため、RPC の話では「クライアント」と「サーバー」が曖昧になることがあります。「サーバー」は Durable Object または WorkerEntrypoint、「クライアント」はバインディング経由でサーバーを呼び出した Worker です。ただし RPC は双方向に流れます。個別の RPC について話すときは、「呼び出し元」と「呼び出し先」を使うことを推奨します。

自分で定義したクラスのインスタンスを RPC メソッドのパラメーターまたは戻り値に使うには、組み込みの RpcTarget クラスを拡張する必要があります。

次の例を考えてください。

{
	"$schema": "./node_modules/wrangler/config-schema.json",
	"name": "counter",
	"main": "./src/counter.js"
}

"$schema" = "./node_modules/wrangler/config-schema.json"
name = "counter"
main = "./src/counter.js"

import { WorkerEntrypoint, RpcTarget } from "cloudflare:workers";

class Counter extends RpcTarget {
	#value = 0;

	increment(amount) {
		this.#value += amount;
		return this.#value;
	}

	get value() {
		return this.#value;
	}
}

export class CounterService extends WorkerEntrypoint {
	async newCounter() {
		return new Counter();
	}
}

export default {
	fetch() {
		return new Response("ok");
	},
};

import { WorkerEntrypoint, RpcTarget } from "cloudflare:workers";

class Counter extends RpcTarget {
	#value = 0;

	increment(amount: number) {
		this.#value += amount;
		return this.#value;
	}

	get value() {
		return this.#value;
	}
}

export class CounterService extends WorkerEntrypoint {
	async newCounter() {
		return new Counter();
	}
}

export default {
	fetch() {
		return new Response("ok");
	},
};

メソッド increment はクライアントから直接呼べます。公開プロパティ value も同様です。

{
	"$schema": "./node_modules/wrangler/config-schema.json",
	"name": "client-worker",
	"main": "./src/clientWorker.js",
	"services": [
		{
			"binding": "COUNTER_SERVICE",
			"service": "counter",
			"entrypoint": "CounterService"
		}
	]
}

"$schema" = "./node_modules/wrangler/config-schema.json"
name = "client-worker"
main = "./src/clientWorker.js"

[[services]]
binding = "COUNTER_SERVICE"
service = "counter"
entrypoint = "CounterService"

export default {
	async fetch(request, env) {
		using counter = await env.COUNTER_SERVICE.newCounter();

		await counter.increment(2); // returns 2
		await counter.increment(1); // returns 3
		await counter.increment(-5); // returns -2

		const count = await counter.value; // returns -2

		return new Response(count);
	},
};

export default {
	async fetch(request: Request, env: Env) {
		using counter = await env.COUNTER_SERVICE.newCounter();

		await counter.increment(2); // returns 2
		await counter.increment(1); // returns 3
		await counter.increment(-5); // returns -2

		const count = await counter.value; // returns -2

		return new Response(count);
	},
};

from workers import WorkerEntrypoint, Response

class Default(WorkerEntrypoint):
    async def fetch(self, request):
        counter = await self.env.COUNTER_SERVICE.new_counter()

        await counter.increment(2)  # returns 2
        await counter.increment(1)  # returns 3
        await counter.increment(-5)  # returns -2

        count = await counter.value  # returns -2

        return Response(str(count))

RpcTarget を拡張するクラスは関数に近い動きをします。オブジェクト自体はシリアライズされず、スタブに置き換わります。この場合、スタブ自体は呼び出せませんが、そのメソッドは呼び出せます。スタブ上の任意のメソッドを呼ぶと、元のオブジェクトが作られた場所へ RPC が戻ります。

上で示したとおり、クラスのプロパティにもアクセスできます。プロパティは引数のない RPC メソッドのように振る舞います。プロパティを await すると、現在値を非同期に取得します。プロパティを取得するきっかけは、await すること（裏では .then() が呼ばれること）です。await せずにプロパティへアクセスしても、取得は走りません。

RPC メソッドを呼んでオブジェクトを受け取ったあと、すぐにそのオブジェクトのメソッドを呼ぶことはよくあります。

// Two round trips.
using counter = await env.COUNTER_SERVICE.getCounter();
await counter.increment();

// Two round trips.
using counter = await env.COUNTER_SERVICE.getCounter();
await counter.increment();

# Two round trips.
counter = await self.env.COUNTER_SERVICE.get_counter()
await counter.increment()

ただし、呼ぶ先の Worker サービスがネットワーク的に遠い場合もあります。Smart Placement や Durable Objects がその例です。上のコードは、getCounter() の呼び出しと .increment() の呼び出しで往復が 2 回になります。これは避けたいです。

多くの RPC システムでは、2 つの呼び出しを getCounterAndIncrement() のような 1 つの「バッチ」呼び出しにまとめるしかありません。ただしインターフェイスは悪化します。ローカルのインターフェイスをこうは設計しません。

Workers RPC では別の方法が使えます。最初の await を省略するだけです。

// Only one round trip! Note the missing `await`.
using promiseForCounter = env.COUNTER_SERVICE.getCounter();
await promiseForCounter.increment();

// Only one round trip! Note the missing `await`.
using promiseForCounter = env.COUNTER_SERVICE.getCounter();
await promiseForCounter.increment();

# Only one round trip! Note the missing await.
promise_for_counter = self.env.COUNTER_SERVICE.get_counter()
await promise_for_counter.increment()

このコードでは、getCounter() はカウンターの Promise を返します。通常、Promise に対して行うことは await だけです。ただし Workers RPC の Promise は特別です。将来の結果に対する投機的な呼び出しを開始できます。これらの呼び出しは、最初の呼び出しの完了を待たずにすぐサーバーへ送られます。そのため、連鎖した複数の呼び出しを 1 回の往復で完了できます。

仕組みはどうなっているのでしょうか。RPC が返す Promise は、本物の JavaScript Promise ではありません。カスタムな "Thenable" ↗ です。Promise と同様の .then() メソッドがあり、通常の Promise を使う場所で使えます。たとえば await できます。それに加えて、RPC Promise はスタブとしても振る舞います。Promise 上の任意のメソッド名を呼ぶと、その Promise の最終結果に対する投機的な呼び出しになります。これが「Promise パイプライン」です。

RPC メソッドが返すオブジェクトのプロパティを呼ぶときも同様です。例:

import { WorkerEntrypoint } from "cloudflare:workers";

export class MyService extends WorkerEntrypoint {
	async foo() {
		return {
			bar: {
				baz: () => "qux",
			},
		};
	}
}

import { WorkerEntrypoint } from "cloudflare:workers";

export class MyService extends WorkerEntrypoint {
	async foo() {
		return {
			bar: {
				baz: () => "qux",
			},
		};
	}
}

from workers import WorkerEntrypoint

class MyService(WorkerEntrypoint):
    async def foo(self):
        return {
            "bar": {
                "baz": lambda: "qux",
            },
        }

export default {
	async fetch(request, env) {
		using foo = env.MY_SERVICE.foo();
		let baz = await foo.bar.baz();
		return new Response(baz);
	},
};

export default {
	async fetch(request, env) {
		using foo = env.MY_SERVICE.foo();
		let baz = await foo.bar.baz();
		return new Response(baz);
	},
};

from workers import WorkerEntrypoint, Response

class Default(WorkerEntrypoint):
    async def fetch(self, request):
        foo = self.env.MY_SERVICE.foo()
        baz = await foo.bar.baz()
        return Response(baz)

最初の RPC が例外をスローした場合、パイプラインされた呼び出しも同じ例外で失敗します。

ReadableStream、WritableStream、Request、Response は、RPC メソッドで送受信できます。そのとき、本文のバイトは適切なフロー制御で自動的にストリーミングされます。通常の 32 MiB 上限 より大きいメッセージを RPC で送れます。

サポートされるのは バイト指向ストリーム ↗（基盤のバイトソースが type: "bytes" のストリーム）だけです。

いずれの場合も、ストリームの所有権は受け手へ移ります。送ったあと、送り手はストリームを読み書きできません。送り手が自分用のコピーを残したい場合は、ReadableStream の tee() メソッド ↗ または Request / Response の clone() メソッド ↗ を使えます。ただし、システムがバイトをバッファし、フロー制御の利点を失うことがあります。

ある Worker から RPC で受け取ったスタブを、別の Worker へ RPC で転送できます。

using counter = env.COUNTER_SERVICE.getCounter();
await env.ANOTHER_SERVICE.useCounter(counter);

using counter = env.COUNTER_SERVICE.getCounter();
await env.ANOTHER_SERVICE.useCounter(counter);

counter = self.env.COUNTER_SERVICE.get_counter()
await self.env.ANOTHER_SERVICE.use_counter(counter)

ここでは 3 つの異なる Workers が関わります。

1. 呼び出し元の Worker（ここでは「紹介者」と呼びます）
2. COUNTER_SERVICE
3. ANOTHER_SERVICE

ANOTHER_SERVICE が渡された counter のメソッドを呼ぶと、この呼び出しは自動で紹介者を経由し、COUNTER_SERVICE が実装する RpcTarget クラスへ届きます。

こうして紹介者 Worker は、もともと直接接続できない 2 つの Workers をつなげられます。

現在、このプロキシは Workers の実行コンテキストが終わるまでだけ続きます。プロキシ接続を後で使うために永続化することはできません。

この動画では、Cloudflare Workers が Remote Procedure Calls（RPC）をサポートし、Workers 間の通信をどう簡単にするかを解説します。JavaScript アプリへの RPC の実装と、サーバーレス構成の作り方を学びます。マイクロサービスの管理や Web アーキテクチャの最適化でも、Cloudflare Workers で RPC をすばやく設定・利用する方法を示します。この動画の最後には、Workers 間での関数呼び出し、引数としての関数の受け渡し、Cloudflare Workers でのユーザー認証の実装が分かるようになります。

• ライフサイクル
• 予約済みメソッド
• 可視性とセキュリティモデル
• TypeScript
• エラーハンドリング

• RPC 呼び出しでは、現在 Smart Placement は無視されます。Worker A で Smart Placement が有効で、Worker B がそれへの Service Binding を宣言している場合、Worker B が RPC で Worker A を呼ぶと、Worker A は同じマシン上でローカルに実行されます。

• シリアライズされた RPC の上限は 32 MiB です。より多くのデータを返す場合は ReadableStream の利用を検討してください。

  export class MyService extends WorkerEntrypoint {
  	async foo() {
  		// Although this works, it puts a lot of memory pressure on the isolate.
  		// If possible, streaming the data from its original source is much preferred and would yield better performance.
  		// If you must buffer the data into memory, consider chunking it into smaller pieces if possible.
  
  		const sizeInBytes = 33 * 1024 * 1024; // 33 MiB
  		const arr = new Uint8Array(sizeInBytes);
  
  		return new ReadableStream({
  			start(controller) {
  				controller.enqueue(arr);
  				controller.close();
  			},
  		});
  	}
  }

  export class MyService extends WorkerEntrypoint {
  	async foo() {
  		// Although this works, it puts a lot of memory pressure on the isolate.
  		// If possible, streaming the data from its original source is much preferred and would yield better performance.
  		// If you must buffer the data into memory, consider chunking it into smaller pieces if possible.
  
  		const sizeInBytes = 33 * 1024 * 1024; // 33 MiB
  		const arr = new Uint8Array(sizeInBytes);
  
  		return new ReadableStream({
  			start(controller) {
  				controller.enqueue(arr);
  				controller.close();
  			},
  		});
  	}
  }