Diffinery

codician/diffinerypublic

Open-source, agent-first GitHub and GitHub Actions alternative that runs on your own Cloudflare account: git hosting, pull requests, code review and drop-in GitHub Actions YAML on Workers, Durable Objects, R2 and Containers.

Artifact and cache uploads: join blocks with native pipes, so large block lists no longer fail or stall (#232)

* Artifact and cache uploads: join blocks with native pipes, so a large block list no longer fails or stalls
test-bot authored 1 parent 76ac40ccommit c0b2e17785ad85018c5b81e03a0edb4129a51300Browse files
3 files changed+32−28
modifiedapps/diffinery/src/http/actions-runtime.ts+1−0View file
@@ -464,6 +464,7 @@
464464 return created(head?.httpEtag ?? `"${newId()}"`);
465465 } catch (e) {
466466 const { status, message } = failure(e);
467+ if (status >= 500) console.error("blob upload failed", b.kind, b.id, q.get("comp") ?? "put", e);
467468 return azureError(status, status === 404 ? "ResourceNotFound" : status === 413 ? "RequestBodyTooLarge" : status === 409 ? "BlobImmutable" : status < 500 ? "InvalidInput" : "InternalError", message);
468469 }
469470 });
modifiedapps/diffinery/src/lib/blob-store.ts+23−28View file
@@ -306,44 +306,39 @@
306306 return out;
307307 }
308308
309-/** The concatenation of `slices`, read one after another. */
310-function concat(bucket: BlobBucket, slices: Slice[]): ReadableStream {
311- let i = 0;
312- let reader: ReadableStreamDefaultReader<Uint8Array> | null = null;
313- return new ReadableStream<Uint8Array>({
314- async pull(ctl) {
315- for (;;) {
316- if (!reader) {
317- const s = slices[i++];
318- if (!s) return ctl.close();
319- const o = await bucket.get(s.key, s.whole ? undefined : { range: { offset: s.offset, length: s.length } });
320- if (!o) throw new Error(`staged block ${s.key} is missing`);
321- reader = o.body.getReader() as ReadableStreamDefaultReader<Uint8Array>;
322- }
323- const { done, value } = await reader.read();
324- if (done) {
325- reader = null;
326- continue;
327- }
328- ctl.enqueue(value);
329- return;
309+/**
310+ * The concatenation of `slices` (`length` bytes), read one after another. Each slice's R2 body is
311+ * piped into one fixed-length stream, which the runtime does natively: a JavaScript loop over the
312+ * chunks kept a Durable Object's CPU busy, so eight steps at once ran little faster than one, and
313+ * the Durable Object answered some of them with errors when it was overloaded.
314+ */
315+function concat(bucket: BlobBucket, slices: Slice[], length: number): ReadableStream {
316+ const F = (globalThis as { FixedLengthStream?: new (n: number) => TransformStream }).FixedLengthStream;
317+ const t: TransformStream = F ? new F(length) : new TransformStream();
318+ void (async () => {
319+ try {
320+ for (const s of slices) {
321+ const o = await bucket.get(s.key, s.whole ? undefined : { range: { offset: s.offset, length: s.length } });
322+ if (!o) throw new Error(`staged block ${s.key} is missing`);
323+ await o.body.pipeTo(t.writable, { preventClose: true });
330324 }
331- },
332- async cancel() {
333- await reader?.cancel().catch(() => {});
334- },
335- });
325+ await t.writable.close();
326+ } catch (e) {
327+ await t.writable.abort(e).catch(() => {});
328+ }
329+ })();
330+ return t.readable;
336331 }
337332
338333 /** Runs one step against the bucket. Returns the parts it uploaded. */
339334 export async function runComposeStep(bucket: BlobBucket, step: ComposeStep, contentType?: string): Promise<Array<{ partNumber: number; etag: string }>> {
340335 if (step.kind === "merge") {
341- await bucket.put(step.target, sized(concat(bucket, step.slices), step.length), contentType ? { httpMetadata: { contentType } } : undefined);
336+ await bucket.put(step.target, concat(bucket, step.slices, step.length), contentType ? { httpMetadata: { contentType } } : undefined);
342337 return [];
343338 }
344339 const upload = bucket.resumeMultipartUpload(step.key, step.uploadId);
345340 const out: Array<{ partNumber: number; etag: string }> = [];
346- for (const p of step.parts) out.push(await upload.uploadPart(p.partNumber, sized(concat(bucket, p.slices), p.length)));
341+ for (const p of step.parts) out.push(await upload.uploadPart(p.partNumber, concat(bucket, p.slices, p.length)));
347342 return out;
348343 }
349344
modifiedapps/diffinery/test/unit/actions-runtime.test.ts+8−0View file
@@ -276,6 +276,14 @@
276276 expect((await run([16 * MiB, 1])).multipart).toBe(1);
277277 });
278278
279+ it("fails a step whose staged block went missing, instead of writing a short part", async () => {
280+ const m = memoryBucket();
281+ await m.bucket.put("s/a", filled(1000, 1));
282+ const step: ComposeStep = { kind: "merge", target: "s/out", slices: [{ key: "s/a", offset: 0, length: 1000, whole: true }, { key: "s/gone", offset: 0, length: 10, whole: true }], length: 1010 };
283+ await expect(runComposeStep(m.bucket, step)).rejects.toThrow(/s\/gone is missing/);
284+ expect(m.objects.has("s/out")).toBe(false);
285+ });
286+
279287 it("merges many small blocks in rounds so that no step exceeds its calls", async () => {
280288 // 10 calls per step: 300 blocks need two rounds of merging before the parts
281289 const r = await run(Array.from({ length: 300 }, (_, i) => 40 * 1024 + i), { stepOps: 10, maxMergeBytes: 1 * MiB });