diff --git a/packages/shade-files/src/client/http-client.ts b/packages/shade-files/src/client/http-client.ts index 7114cd8..dca1ae3 100644 --- a/packages/shade-files/src/client/http-client.ts +++ b/packages/shade-files/src/client/http-client.ts @@ -444,8 +444,12 @@ export function createFilesHttpClient( }; return out; } - // Streamed read — only supported when the queue drainer is wired. - if (drainer === null) { + // Streamed read — only supported when pull-mode was configured. + // `drainer` is assigned asynchronously after the streams bridge has + // subscribed, so checking it here races client construction. The + // promise itself is the synchronous configuration signal; awaiting it + // also guarantees the drainer has been installed by its `.then()`. + if (streamsBridgePromise === null) { throw new InternalFileError( `http RPC client received a streamed read (size ${wire.size}) but is in inline-only mode. Pass { outboundQueueUrl, transferBaseUrl } when constructing the client to enable streamed reads.`, ); @@ -507,8 +511,11 @@ export function createFilesHttpClient( ); } - // Streamed write — requires the queue drainer + streams-bridge. - if (drainer === null) { + // Streamed write — requires pull-mode configuration. Do not inspect + // `drainer` as a readiness flag: bridge registration is asynchronous, + // and an immediate large write must wait for it instead of being + // misclassified as an inline-only client. + if (streamsBridgePromise === null) { throw new ConflictError( `http RPC client supports inline writes only (≤ ${INLINE_THRESHOLD} bytes). The supplied input was promoted to streams (size ${decision.size ?? 'unknown'}). Pass { outboundQueueUrl, transferBaseUrl } to enable streamed writes.`, ); diff --git a/packages/shade-files/tests/integration/http-rpc-streams.test.ts b/packages/shade-files/tests/integration/http-rpc-streams.test.ts index f67acb5..7c40b22 100644 --- a/packages/shade-files/tests/integration/http-rpc-streams.test.ts +++ b/packages/shade-files/tests/integration/http-rpc-streams.test.ts @@ -7,8 +7,10 @@ import { } from '@shade/server'; import { SubtleCryptoProvider } from '@shade/crypto-web'; import { Hono } from 'hono'; +import { createFilesHttpClient, type FileEntry } from '../../src/index.js'; const crypto = new SubtleCryptoProvider(); +type ShadeInstance = Awaited>; /** * Stand up the full pull-mode rig: @@ -20,7 +22,8 @@ const crypto = new SubtleCryptoProvider(); * no inbound listener, streams supported via long-poll. */ async function setupPullRig(opts: { - bobHandler: Parameters>['files']>['serve']>[0]; + bobHandler: Parameters['serve']>[0]; + wrapClientShade?: (shade: ShadeInstance) => Parameters[0]; }) { const prekey = createPrekeyServer({ crypto, @@ -46,7 +49,7 @@ async function setupPullRig(opts: { const bobServer = Bun.serve({ port: 0, fetch: app.fetch }); const baseUrl = `http://localhost:${bobServer.port}`; - const fs = alice.files.httpClient('bob', { + const fs = createFilesHttpClient(opts.wrapClientShade?.(alice) ?? alice, 'bob', { rpcUrl: `${baseUrl}/rpc`, outboundQueueUrl: `${baseUrl}/queue`, transferBaseUrl: baseUrl, @@ -70,6 +73,83 @@ async function setupPullRig(opts: { } describe('@shade/files HTTP RPC — pull-mode streams', () => { + test('immediate streamed write waits for asynchronous stream-bridge registration', async () => { + const payload = new Uint8Array(512 * 1024); + for (let i = 0; i < payload.length; i++) payload[i] = (i * 53) & 0xff; + + let releaseRegistration!: () => void; + const registrationGate = new Promise((resolve) => { + releaseRegistration = resolve; + }); + let registrationStarted!: () => void; + const started = new Promise((resolve) => { + registrationStarted = resolve; + }); + + const rig = await setupPullRig({ + wrapClientShade: (alice) => + new Proxy(alice, { + get(target, property) { + if (property === 'onIncomingTransfer') { + return async (handler: Parameters[0]) => { + registrationStarted(); + await registrationGate; + return await target.onIncomingTransfer(handler); + }; + } + const value = Reflect.get(target, property, target) as unknown; + return typeof value === 'function' ? value.bind(target) : value; + }, + }), + bobHandler: { + write: async (ctx) => { + const content = ctx.args.content; + if (content.kind !== 'streams') throw new Error('expected streamed content'); + const reader = content.stream.getReader(); + let received = 0; + while (true) { + const { value, done } = await reader.read(); + if (done) break; + received += value?.byteLength ?? 0; + } + reader.releaseLock(); + await content.sha256; + const entry: FileEntry = { + name: 'immediate.bin', + kind: 'file', + size: received, + mtime: Date.now(), + metadata: {}, + }; + return { entry }; + }, + }, + }); + + try { + let settled = false; + const write = rig.fs.write('/immediate.bin', payload); + void write.then( + () => { + settled = true; + }, + () => { + settled = true; + }, + ); + await started; + await Bun.sleep(10); + expect(settled).toBe(false); + + releaseRegistration(); + const result = await write; + expect(result.entry.size).toBe(payload.byteLength); + } finally { + releaseRegistration(); + await rig.teardown(); + } + }, 15_000); + test('streamed read (4 MiB) via long-poll queue', async () => { const payload = new Uint8Array(4 * 1024 * 1024); for (let i = 0; i < payload.length; i++) payload[i] = (i * 97) & 0xff;