Function transportFailingBackend

packages/computer/src/workspace.test.ts:1165–1221

(
    id: string,
    onConnect: () => void,
  )

  // holding a dead handle. The cache must drop the handle on the
  // way out so the next operation reconnects.
  function transportFailingBackend(
    id: string,
    onConnect: () => void,
  ): { backend: WorkspaceBackend; failNext: () => void } {
    let shouldFail = false;
    const sync: import("@cloudflare/computer-rpc").SyncRPC = {
      ...fakeRpc(),
      async push(input) {
        if (shouldFail) throw new WorkspaceTransportError("WebSocket closed");
        const reader = input.changes.getReader();
        try {
          while (true) {
            const { done } = await reader.read();
            if (done) break;
          }
        } finally {
          reader.releaseLock();
        }
        return { rev: 0, appliedPushCursor: { rev: input.senderRev, path: null } };
      },
      async fetchChanges() {
        if (shouldFail) throw new WorkspaceTransportError("WebSocket closed");
        return {
          currentCursor: { rev: 0, path: null },
          appliedPushCursor: { rev: 0, path: null },
          stream: new ReadableStream<import("@cloudflare/dofs").ChangeEntry>({
            start(c) {
              c.close();
            },
          }),
        };
      },
      async watermarks() {
        return { currentRev: 0, pushRev: 0, fetchCursor: { rev: 0, path: null } };
      },
    };
    const backend: WorkspaceBackend = {
      id,
      type: "fake",
      async connect(): Promise<BackendHandle> {
        onConnect();
        // closed promise stays pending forever — simulating a
        // wedged transport that never produces a clean signal.
        return {
          rpc: composite(sync),
          closed: new Promise<void>(() => {}),
          close: async () => {},
        };
      },
    };
    return {
      backend,
      failNext: () => {
        shouldFail = true;
      },
    };
  }