[DO] Adding that ReadableStream cancellation propagates from Worker to DO

[Oxyjun]

Summary

Screenshots (optional)

Documentation checklist

• The documentation style guide has been adhered to.
• If a larger change - such as adding a new page- an issue has been opened in relation to any incorrect or out of date information that this PR fixes.
• Files which have changed name or location have been allocated redirects.

[hyperlint-ai]

Howdy and thanks for contributing to our repo. The Cloudflare team reviews new, external PRs within two (2) weeks. If it's been two weeks or longer without any movement, please tag the PR Assignees in a comment.

We review internal PRs within 1 week. If it's something urgent or has been sitting without a comment, start a thread in the Developer Docs space internally.

---

PR Change Summary

Enhanced documentation to clarify the propagation of ReadableStream cancellation from Workers to Durable Objects.

• Added a section on ReadableStream cancellation in relation to Durable Objects.
• Clarified the behavior of cancellation propagation in the documentation.

Modified Files

• src/content/docs/workers/runtime-apis/streams/readablestream.mdx

---

How can I customize these reviews?

Check out the Hyperlint AI Reviewer docs for more information on how to customize the review.

If you just want to ignore it on this PR, you can add the hyperlint-ignore label to the PR. Future changes won't trigger a Hyperlint review.

Note specifically for link checks, we only check the first 30 links in a file and we cache the results for several hours (for instance, if you just added a page, you might experience this). Our recommendation is to add hyperlint-ignore to the PR to ignore the link check for this PR.

[github-actions]

This pull request requires reviews from CODEOWNERS as it changes files that match the following patterns:

Pattern	Owners
/src/content/docs/durable-objects/	@elithrar, @vy-ton, @joshthoward, @oxyjun, @harshil1712, @mikenomitch, @cloudflare/pcx-technical-writing, @cloudflare/workers-runtime-1
/src/content/docs/workers/	@cloudflare/workers-docs, @GregBrimble, @irvinebroque, @mikenomitch, @korinne, @WalshyDev, @cloudflare/deploy-config, @cloudflare/pcx-technical-writing, @kodster28, @cloudflare/wrangler, @cloudflare/workers-runtime-1, @cloudflare/wrangler

[github-actions]

Preview URL: https://23281253.preview.developers.cloudflare.com
Preview Branch URL: https://jun-do-readable-stream-propagation.preview.developers.cloudflare.com

Files with changes (up to 15)

Original Link	Updated Link
https://developers.cloudflare.com/durable-objects/examples/readable-stream/	https://jun-do-readable-stream-propagation.preview.developers.cloudflare.com/durable-objects/examples/readable-stream/
https://developers.cloudflare.com/workers/runtime-apis/streams/readablestream/	https://jun-do-readable-stream-propagation.preview.developers.cloudflare.com/workers/runtime-apis/streams/readablestream/

[TimoWilhelm]

This should return the values and simplify the cancellation a bit

import { DurableObject } from 'cloudflare:workers';

async function* dataSource(signal: AbortSignal) {
    let counter = 0;
    while (!signal.aborted) {
        yield counter++;
        await new Promise((resolve) => setTimeout(resolve, 1_000));
    }

    console.log('Data source cancelled');
}

export class MyDurableObject extends DurableObject<Env> {
    async fetch(request: Request): Promise<Response> {
        const abortController = new AbortController();

        const stream = new ReadableStream({
            async start(controller) {
                if (request.signal.aborted) {
                    controller.close();
                    abortController.abort();
                    return;
                }

                for await (const value of dataSource(abortController.signal)) {
                    controller.enqueue(new TextEncoder().encode(String(value)));
                }
            },
            cancel() {
                console.log('Stream cancelled');
                abortController.abort();
            },
        });

        const headers = new Headers({
            'Content-Type': 'application/octet-stream',
        });

        return new Response(stream, { headers });
    }
}

export default {
    async fetch(request, env, ctx): Promise<Response> {
        const id: DurableObjectId = env.MY_DURABLE_OBJECT.idFromName('my-id');
        const stub = env.MY_DURABLE_OBJECT.get(id);
        const response = await stub.fetch(request, { ...request });
        if (!response.ok || !response.body) {
            return new Response('Invalid response', { status: 500 });
        }

        const reader = response.body.pipeThrough(new TextDecoderStream()).getReader();

        let data = [] as string[];
        let i = 0;
        while (true) {
            // Cancel the stream after 5 messages
            if (i > 5) {
                reader.cancel();
                break;
            }
            const { value, done } = await reader.read();

            if (value) {
                console.log(`Got value ${value}`);
                data = [...data, value];
            }

            if (done) {
                break;
            }
            i++;
        }

        return Response.json(data);
    },
} satisfies ExportedHandler<Env>;