はじめに
通常の HTTP リクエストは、リクエストを開始し、1 つのレスポンスを受け取って終了します。WebSocket は最初の HTTP リクエストを、開いたまま維持される双方向接続にアップグレードします。そのため、サーバーは何かが変化した時点で更新を送信できます。チャットメッセージ、共同編集用のカーソル、リアルタイムの注文ボードなどは、すべてこのリアルタイム通信の恩恵を受けます。
Durable Object は、各ルームに 1 つの調整ポイントを提供します。入口となる Worker は、planning のように検証済みのルーム名を安定したオブジェクト ID に変換します。選択されたオブジェクトは、そのルームの WebSocket 接続を受け付け、受信したすべてのメッセージを検証し、承認した更新だけを自身の接続中クライアントにブロードキャストします。別の名前を指定すると別のオブジェクトが選択されるため、support が planning の通信を受け取ることはありません。
この実験では、標準の WebSocket API を使用し、アクティブなソケットの集合をメモリ上に保持します。これにより、O06 で WebSocket Hibernation と接続アタッチメントを導入する前に、接続とブロードキャストの動作を確認できます。SQLite には少量のメッセージ履歴を保存します。これにより、不正な入力が永続状態を変更していないことを確認できます。ただし、開いたソケット自体を永続化するわけではありません。
この実験では、プロトコルを実装し、用意された 2 つのクライアントを 1 つのルームに接続し、3 つ目のクライアントを別のルームに接続します。その後、正しい更新のブロードキャストを確認し、不正な入力を拒否し、Cloudflare 上でテストを繰り返します。さらに、ブラウザクライアントと Dashboard を確認し、最後に使い捨てリソースを正確に削除します。
新しい VM では、それぞれ Wrangler の認証が必要です。O01〜O04 で扱った、安定した Durable Object 名、バインディング、RPC、SQLite を利用した状態管理について理解していることを前提とします。セットアップでは、Node.js 22.22.0、プロジェクトローカルの Wrangler 4.132.0、および /home/labex/project/room-broadcast に ws テストクライアントをインストールします。ブラウザクライアントとテストクライアントは用意されますが、Worker の作成、Cloudflare の認証、デプロイは行いません。
VM を認証し、ルーム名前空間を宣言する
このステップでは、新しい VM を認証し、リアルタイムルーム用に SQLite ベースの Durable Object クラスを 1 つ宣言します。
用意されたプロジェクトに移動し、固定された Wrangler のバージョンを確認して、この VM を認証します。
cd /home/labex/project/room-broadcast
npx wrangler --version
npx wrangler login --device --browser=false
Wrangler のバージョンとして 4.132.0 が表示されることを確認します。表示された Cloudflare URL をブラウザで開き、短いコードを入力し、使用する学習用アカウントを確認して認証します。ブラウザが Wrangler にアクセス権を付与します。パスワードが VM に送信されることはありません。
安全な ID 情報だけを読み取り、確認したアカウントを選択して、使い捨て Worker 用の一意な名前を生成します。
WHOAMI="$(npx wrangler whoami --json)"
printf '%s\n' "$WHOAMI" | jq '{loggedIn, authType, accounts: [.accounts[] | {name}]}'
ACCOUNT_ID="$(printf '%s\n' "$WHOAMI" | jq -r '.accounts[] | select(.name == "LabEx Learning") | .id')"
test -n "$ACCOUNT_ID"
RUN="labex-c10-o05-$(openssl rand -hex 6)"
printf '%s\n' "$RUN" | tee .labex/run-name
専用の学習用アカウントに別の表示名がある場合は、確認した名前に置き換えてください。続けて設定を書き込みます。
cat > wrangler.jsonc <<JSON
{
"\$schema": "./node_modules/wrangler/config-schema.json",
"name": "$RUN",
"account_id": "$ACCOUNT_ID",
"main": "src/index.js",
"compatibility_date": "2026-09-18",
"workers_dev": true,
"preview_urls": false,
"observability": { "enabled": true, "head_sampling_rate": 1 },
"durable_objects": { "bindings": [
{ "name": "ROOMS", "class_name": "RoomBroadcast" }
] },
"exports": {
"RoomBroadcast": { "type": "durable-object", "storage": "sqlite" }
}
}
JSON
ROOMS バインディングは、Worker からクラス名前空間へアクセスするための経路です。getByName("planning") を呼び出すと常に同じ論理ルームが選択され、getByName("support") を呼び出すと独立したオブジェクトが選択されます。エクスポートにより、選択された各ルームが専用の SQLite ストレージを持ちます。デプロイするまで、クラウドリソースは作成されません。
検証済み WebSocket プロトコルを実装する
このステップでは、簡単なメッセージ契約を定義し、WebSocket メッセージを受け付けてブロードキャストするルームオブジェクトを実装します。
最初のリクエストには Upgrade: websocket が含まれている必要があります。アップグレード後、メッセージは新しい HTTP リクエストではなくフレームとして送信されます。クライアントはフレーム内に任意のテキストを送信できるため、JSON の解析は最初の確認にすぎません。永続状態を変更する前に、想定された type、空でない長さ制限付きの text フィールド、そして想定外のフィールドが存在しないことも検証する必要があります。
共有プロトコル用のヘルパーを作成します。
cat > src/protocol.js <<'JS'
const ROOM_PATTERN = /^[a-z0-9](?:[a-z0-9-]{0,38}[a-z0-9])?$/;
export function parseRoomPath(pathname) {
const match = pathname.match(/^\/rooms\/([^/]+)\/(connect|state)$/);
if (!match || !ROOM_PATTERN.test(match[1])) return null;
return { room: match[1], action: match[2] };
}
export function parseClientMessage(raw) {
if (typeof raw !== "string" || raw.length > 512) return { ok: false };
let value;
try { value = JSON.parse(raw); } catch { return { ok: false }; }
if (!value || typeof value !== "object" || Array.isArray(value)) return { ok: false };
const keys = Object.keys(value).sort();
if (keys.length !== 2 || keys[0] !== "text" || keys[1] !== "type") return { ok: false };
if (value.type !== "update" || typeof value.text !== "string") return { ok: false };
const text = value.text.trim();
if (text.length < 1 || text.length > 80) return { ok: false };
return { ok: true, text };
}
JS
入口となる Worker と Durable Object クラスを作成します。
cat > src/index.js <<'JS'
import { DurableObject } from "cloudflare:workers";
import { CLIENT_HTML } from "./client-html.js";
import { parseClientMessage, parseRoomPath } from "./protocol.js";
const json = (body, status = 200) => Response.json(body, { status });
export class RoomBroadcast extends DurableObject {
constructor(ctx, env) {
super(ctx, env);
this.sessions = new Set();
this.ctx.blockConcurrencyWhile(async () => {
this.ctx.storage.sql.exec(`
CREATE TABLE IF NOT EXISTS messages (
sequence INTEGER PRIMARY KEY AUTOINCREMENT,
text TEXT NOT NULL,
created_at INTEGER NOT NULL
)
`);
});
}
async fetch(request) {
if ((request.headers.get("Upgrade") || "").toLowerCase() !== "websocket") {
return json({ error: "websocket_upgrade_required" }, 426);
}
const pair = new WebSocketPair();
const [client, server] = Object.values(pair);
server.accept();
this.sessions.add(server);
server.addEventListener("message", event => this.receive(server, event.data));
const forget = () => this.sessions.delete(server);
server.addEventListener("close", forget);
server.addEventListener("error", forget);
server.send(JSON.stringify({ type: "ready" }));
return new Response(null, { status: 101, webSocket: client });
}
receive(sender, raw) {
const message = parseClientMessage(raw);
if (!message.ok) {
sender.send(JSON.stringify({
type: "error",
code: "invalid_message",
detail: "Send only {type: update, text: 1-80 characters}."
}));
return;
}
const createdAt = Date.now();
const row = this.ctx.storage.sql.exec(`
INSERT INTO messages (text, created_at)
VALUES (?, ?)
RETURNING sequence
`, message.text, createdAt).one();
const update = JSON.stringify({
type: "update",
sequence: row.sequence,
text: message.text,
createdAt
});
for (const socket of this.sessions) {
try { socket.send(update); } catch { this.sessions.delete(socket); }
}
console.log(JSON.stringify({ event: "room_update", sequence: row.sequence, connected: this.sessions.size }));
}
async getState() {
const messages = this.ctx.storage.sql.exec(`
SELECT sequence, text, created_at AS createdAt
FROM messages ORDER BY sequence
`).toArray();
return {
messageCount: messages.length,
latestSequence: messages.at(-1)?.sequence ?? 0,
messages
};
}
}
export default {
async fetch(request, env) {
const url = new URL(request.url);
if (url.pathname === "/" && request.method === "GET") {
return new Response(CLIENT_HTML, { headers: { "content-type": "text/html; charset=utf-8" } });
}
const route = parseRoomPath(url.pathname);
if (!route) return json({ error: "not_found" }, 404);
if (route.action === "connect") {
if (request.method !== "GET" || (request.headers.get("Upgrade") || "").toLowerCase() !== "websocket") {
return json({ error: "websocket_upgrade_required" }, 426);
}
return env.ROOMS.getByName(route.room).fetch(request);
}
if (request.method !== "GET") return json({ error: "method_not_allowed" }, 405);
const state = await env.ROOMS.getByName(route.room).getState();
return json({ room: route.room, ...state });
}
};
JS
WebSocketPair は、1 つの接続におけるクライアント側とサーバー側の端点を作成します。HTTP 101 とともにクライアント側の端点を返すとアップグレードが完了し、server.accept() によって標準のサーバー側ソケットが開始されます。メモリ上の sessions セットは意図的に 1 つのオブジェクトインスタンス内だけで有効です。安定したルーム名によって、ルームをまたいでこのセットが共有されることを防ぎます。
決定的なプロトコルテストを実行し、デプロイせずに Wrangler でビルドを確認します。
npm test
npx wrangler deploy --dry-run
4 つのテストが成功することを確認します。ドライランでは Worker モジュールとバインディング設定を確認します。実際のソケット動作は、後のライブ手順で検証します。
1 つのルーム内で更新をブロードキャストする
このステップでは、Worker をローカルで実行し、同じルームを共有する 2 つのクライアントには 1 つの更新が届き、別のルームのクライアントには届かないことを確認します。
Wrangler をバックグラウンドジョブとして起動します。出力をリダイレクトするとターミナルを読みやすく保てます。また、保存したジョブ ID を使うことで、後から対象のプロセスだけを停止できます。
mkdir -p .labex/local-state
npx wrangler dev --local --ip 127.0.0.1 --port 8787 --persist-to .labex/local-state > .labex/dev.log 2>&1 &
echo $! > .labex/dev.pid
for attempt in $(seq 1 30); do
curl --silent --fail http://127.0.0.1:8787/ >/dev/null && break
sleep 1
done
curl --silent --fail http://127.0.0.1:8787/ | grep -o '<title>[^<]*</title>'
用意されたクライアントプログラムは、実際の WebSocket 接続を 3 つ開きます。planning に 2 つ、support に 1 つ接続します。最初の planning クライアントから 1 つの更新を送信し、3 つすべてから限定時間内に得られる結果を待ちます。
node tools/room-clients.mjs http://127.0.0.1:8787 planning support broadcast | tee .labex/local-broadcast.json
sender オブジェクトと peer オブジェクトに、同じ sequence: 1 とテキストが含まれていることを確認します。otherUpdates は 0 である必要があります。state セクションには、永続化された planning のメッセージが 1 件、support のメッセージが 0 件あることが個別に表示されます。これにより、設計の 2 つの側面を確認できます。安定した同じ名前によって最初の 2 クライアントが同じルームに参加し、異なる名前によって 3 つ目のクライアントがブロードキャスト範囲の外に置かれます。
状態を変更する前に不正なメッセージを拒否する
このステップでは、JSON としては有効ですがアプリケーション入力としては不正なフレームを送信し、その前後で永続状態を比較します。
重要なのは空の text フィールドです。JSON の解析は成功しますが、ルームプロトコルによって拒否されます。同じローカルオブジェクトに対して、用意された 2 つ目のフェーズを実行します。
node tools/room-clients.mjs http://127.0.0.1:8787 planning support invalid | tee .labex/local-invalid.json
エラーコード invalid_message を受け取るのは送信元クライアントだけです。peerErrors は 0 のままです。before と after の履歴は、メッセージ 1 件を含む同じ内容になります。したがって、不正なクライアントが行を追加したり、シーケンスを進めたり、エラーをルーム全体へのブロードキャストに変えたりすることはありません。
2 つのルームの状態を直接読み取ります。
curl --silent --fail http://127.0.0.1:8787/rooms/planning/state | jq
curl --silent --fail http://127.0.0.1:8787/rooms/support/state | jq
最初のレスポンスにはメッセージが 1 件、2 つ目のレスポンスには 0 件と表示されます。クライアントがテスト後に切断しても、HTTP による状態の読み取り結果が正しい状態を示します。
デプロイしてクラウド上の WebSocket クライアントを実行する
このステップでは、ローカルランタイムを停止し、同じコードをデプロイしたうえで、Cloudflare 経由で 3 クライアントの契約を再確認します。
先ほど保存したローカルジョブだけを停止してから、デプロイします。
kill "$(cat .labex/dev.pid)"
wait "$(cat .labex/dev.pid)" 2>/dev/null || true
npx wrangler deploy | tee .labex/deploy.log
APP_URL="$(grep -Eo 'https://[^ ]+\.workers\.dev' .labex/deploy.log | tail -1)"
test -n "$APP_URL"
printf '%s\n' "$APP_URL" | tee .labex/app-url
デプロイによって、まず Worker が作成され、RoomBroadcast 名前空間が調整されます。ホームページが正常に表示されても、状態を持つルートの準備が完了したことを単独で証明することはできません。そのため、空のルームの状態を読み取って正確な JSON 契約を確認し、短い待機時間を設けます。
for attempt in $(seq 1 30); do
READY="$(curl --silent --show-error "$APP_URL/rooms/cloud-observer/state" || true)"
test "$(printf '%s' "$READY" | jq -r '.messageCount // -1' 2>/dev/null)" = 0 && break
sleep 2
done
test "$(printf '%s' "$READY" | jq -r .messageCount)" = 0
sleep 5
同じ実際の WebSocket クライアントを、一意なクラウドルームに対して実行します。
node tools/room-clients.mjs "$APP_URL" cloud-planning cloud-support broadcast | tee .labex/cloud-broadcast.json
node tools/room-clients.mjs "$APP_URL" cloud-planning cloud-support invalid | tee .labex/cloud-invalid.json
クラウドの出力がローカル開発と同じ動作を示すことを確認します。2 つの planning クライアントがシーケンス 1 を受信し、support クライアントは更新を受信しません。また、不正な入力によって履歴が変更されないことを確認します。
表示された APP_URL をブラウザで開きます。Connect three clients を選択し、続けて Send planning update を選択します。クライアント A と B に同じ新しい update が表示され、クライアント C には ready メッセージだけが表示されます。次に Send malformed update を選択し、エラーがクライアント A だけに表示されることを確認します。確認が終わったら Disconnect clients を選択し、3 つすべてのカードが Closed と表示されるまで待ちます。これにより、ページを離れる前に WebSocket の切断ハンドシェイクが完了します。このページは用意された確認用クライアントです。Node プローブとバックエンドのチェックが、引き続き正式な受け入れ確認となります。
ブラウザクライアントと Durable Object を確認する
このステップでは、実行時の証拠を Cloudflare Dashboard に結び付け、コードを変更せずに再デプロイしても永続的なルーム履歴が残ることを確認します。
ブラウザのデモは、3 つのカードを確認できる時間だけ接続したままにします。2 つの planning カードは、ルーム単位のブロードキャストを示す目に見える証拠です。何も表示されない support カードも同じように重要です。これは、ルームの ID 境界を越えて通信が流れていないことを示します。
Cloudflare Dashboard で Workers & Pages を開き、.labex/run-name に保存されている正確な名前を選択して、バインディングを確認します。ROOMS が RoomBroadcast を指していることを確認してください。次に Durable Objects を開き、<your-worker>_RoomBroadcast という名前の名前空間を選択し、Overview に Storage: SQL と表示されることを確認します。
名前空間の Logs タブを開きます。ブラウザまたは Node プローブに関連する、最近の成功行を選択します。構造化された room_update アプリケーションメッセージには、メッセージ本文を記録せずに、シーケンスと現在の接続数が表示されます。Dashboard へのログ配信には遅延が発生する場合があります。実行時のレスポンスと独立したチェック結果が、引き続き正式な確認結果です。
コードを変更せずに再デプロイします。開いている WebSocket 接続はライブの通信経路であり、デプロイ後も維持されることは保証されません。ただし、SQLite の履歴は名前付きオブジェクトに属しているため、残っているはずです。
npx wrangler deploy
APP_URL="$(cat .labex/app-url)"
curl --silent --fail "$APP_URL/rooms/cloud-planning/state" | jq
curl --silent --fail "$APP_URL/rooms/cloud-support/state" | jq
planning ルームには、シーケンス 1 のメッセージが 1 件残っていることが表示されます。support は空のままです。生成されたサフィックス、タイムスタンプ、Dashboard の通信量合計は、テスト例とは異なります。
ルーム名前空間を削除する
このステップでは、使い捨ての Durable Object 名前空間と Worker を正確に削除し、LabEx が両方のリソースが存在しないことを確認できるよう、VM の認証状態を維持します。
保存した名前が labex-c10-o05- で始まることを確認します。状態を持たないクリーンアップ用エントリポイントを作成します。
RUN="$(cat .labex/run-name)"
case "$RUN" in labex-c10-o05-*) ;; *) echo "Unexpected Worker name" >&2; exit 1;; esac
cat > src/cleanup.js <<'JS'
export default {
fetch() {
return Response.json({ status: "cleanup" }, { status: 410 });
}
};
JS
同じ Worker とアカウントを対象に、クリーンアップ用設定を作成します。state: "deleted" のトゥームストーンによって、この実験のクラス名前空間だけが削除されます。使い捨ての履歴も含まれます。
ACCOUNT_ID="$(node -e 'console.log(JSON.parse(require("fs").readFileSync("wrangler.jsonc", "utf8")).account_id)')"
cat > wrangler.cleanup.jsonc <<JSON
{
"\$schema": "./node_modules/wrangler/config-schema.json",
"name": "$RUN",
"account_id": "$ACCOUNT_ID",
"main": "src/cleanup.js",
"compatibility_date": "2026-09-18",
"workers_dev": true,
"preview_urls": false,
"exports": {
"RoomBroadcast": { "type": "durable-object", "state": "deleted" }
}
}
JSON
npx wrangler deploy --config wrangler.cleanup.jsonc
調整結果に Deleted: RoomBroadcast と表示されることを確認します。残りの状態を持たない Worker を削除します。削除は元に戻せないため、Wrangler は確認を求めます。表示された名前が $RUN の値と完全に一致することを確認してから、削除を承認してください。
npx wrangler delete --config wrangler.cleanup.jsonc
確認プロンプトで y を入力して Enter キーを押します。コマンドが、生成した Worker 名に続く Successfully deleted で終了することを確認します。
このステップの最後に確認できるよう、この VM の認証状態を維持します。Wrangler が認証済みセッションを報告することを確認します。
npx wrangler whoami --json | jq '{loggedIn, authType}'
JSON に "loggedIn": true が含まれている必要があります。これで LabEx は選択したアカウントを照会し、Worker と Durable Object 名前空間の両方が存在しないことを確認できます。ネットワークエラーや認証エラーは、クリーンアップが完了した証拠にはなりません。
この VM の Wrangler 認証を取り消す
このステップでは、クラウドリソースの削除を確認した後、新しい VM にだけ保存されている OAuth 認証を取り消します。
wrangler logout はローカルの認証情報を削除します。通常の人間向け出力は曖昧になる場合があるため、構造化された whoami --json の確認が重要です。loggedIn フィールドが正式な判定結果です。
npx wrangler logout
npx wrangler whoami --json
最後の JSON に "loggedIn": false が含まれている必要があります。これはブラウザ上の Cloudflare 学習用アカウントを削除したり、サインアウトしたりする操作ではありません。この VM から今後認証済み Wrangler リクエストを送信できないようにするだけです。
まとめ
HTTP リクエストを WebSocket にアップグレードし、検証済みのルーム名を独立した Durable Object にルーティングし、同じルームの 2 クライアントに承認済みの更新をブロードキャストしながら、別のルームを分離しました。また、JSON の解析とアプリケーションレベルの検証を分け、不正な入力によってブロードキャスト状態も SQLite の履歴も変更されないことを確認しました。さらに、Cloudflare 上で同じ動作を再現し、ブラウザと Dashboard の表示を確認し、再デプロイ後も履歴が残ることを検証したうえで、使い捨ての名前空間を正確に削除しました。
再利用できる設計上の原則は次のとおりです。状態を選択または変更する前に入力を検証し、各リアルタイムグループを固有の安定したオブジェクト ID で調整し、アクティブな接続と永続的なアプリケーション履歴を別々に扱います。
