diff --git a/src/workerd/api/r2-bucket.c++ b/src/workerd/api/r2-bucket.c++ index 85fd1412c0d..610a2c1f499 100644 --- a/src/workerd/api/r2-bucket.c++ +++ b/src/workerd/api/r2-bucket.c++ @@ -510,6 +510,132 @@ jsg::Promise>> R2Bucket::head(jsg::Lock }); } +jsg::Ref R2Bucket::getRpcMethod(jsg::Lock& js, kj::StringPtr methodName) { + auto fetcher = [&]() -> jsg::Ref { + KJ_SWITCH_ONEOF(clientChannel) { + KJ_CASE_ONEOF(channel, uint) { + return js.alloc( + channel, Fetcher::RequiresHostAndProtocol::NO, true /* isInHouse */); + } + KJ_CASE_ONEOF(channel, IoOwn) { + return js.alloc(IoContext::current().addObject(kj::addRef(*channel)), + Fetcher::RequiresHostAndProtocol::NO, true /* isInHouse */); + } + } + KJ_UNREACHABLE; + }(); + + // getRpcMethodInternal skips the `rpc` compatibility gate, which matters because R2 bindings must + // keep working on compat dates older than that flag. The lookup is lazy and never fails for a + // real method name -- whether the entrypoint actually implements it is only discovered when the + // call reaches the far side. + return KJ_ASSERT_NONNULL(fetcher->getRpcMethodInternal(js, kj::str(methodName))); +} + +namespace { +// Turn the JsRpcPromise a JSRPC call returns into an ordinary jsg::Promise. +// +// JsRpcPromise is a custom thenable whose `then()` takes raw v8 functions and deliberately hides +// the inner promise from JSG, so it cannot be chained from C++ directly. Resolving a fresh promise +// with it makes V8 adopt it, which also keeps this independent of the unwrap_custom_thenables +// compatibility flag. +jsg::Promise normalizeRpcPromise(jsg::Lock& js, jsg::Value rpcPromise) { + auto paf = js.newPromiseAndResolver(); + paf.resolver.resolve(js, kj::mv(rpcPromise)); + return kj::mv(paf.promise); +} +} // namespace + +jsg::Promise>> R2Bucket::headRpc(jsg::Lock& js, + kj::String key, + const jsg::TypeHandler>& rpcPropType, + const jsg::TypeHandler>& fnType, + const jsg::TypeHandler>& resultType) { + return js.evalNow([&] { + auto& context = IoContext::current(); + TraceContext traceContext = context.makeUserTraceSpan("r2_head"_kjc); + + traceContext.setTag("cloudflare.binding.type"_kjc, "r2"_kjc); + KJ_IF_SOME(b, this->bindingName()) { + traceContext.setTag("cloudflare.binding.name"_kjc, b); + } + traceContext.setTag("cloudflare.r2.operation"_kjc, "HeadObject"_kjc); + KJ_IF_SOME(b, this->bucketName()) { + traceContext.setTag("cloudflare.r2.bucket"_kjc, b); + } + traceContext.setTag("cloudflare.r2.request.key"_kjc, key.asPtr()); + + auto rpcProp = getRpcMethod(js, "head"_kj); + auto fn = JSG_REQUIRE_NONNULL(fnType.tryUnwrap(js, rpcPropType.wrap(js, kj::mv(rpcProp))), + Error, "R2 binding entrypoint's head method is not callable"); + + return normalizeRpcPromise(js, fn(js, kj::mv(key))) + .then(js, + [&resultType, traceContext = kj::mv(traceContext)]( + jsg::Lock& js, jsg::Value value) mutable -> kj::Maybe> { + // A missing object is null, not an error: the gateway maps the 404 that + // R2Result::objectNotFound() used to represent onto a null return. + auto parsed = JSG_REQUIRE_NONNULL(resultType.tryUnwrap(js, value.getHandle(js)), Error, + "R2 binding entrypoint returned an unrecognized head result"); + + KJ_IF_SOME(rpc, parsed) { + auto result = js.alloc(kj::mv(rpc.key), kj::mv(rpc.version), rpc.size, + kj::mv(rpc.etag), + js.alloc(kj::mv(rpc.checksums.md5), kj::mv(rpc.checksums.sha1), + kj::mv(rpc.checksums.sha256), kj::mv(rpc.checksums.sha384), + kj::mv(rpc.checksums.sha512)), + rpc.uploaded, + // head always reports http and custom metadata, so an absent field means "none set" + // rather than "not requested". parseObjectMetadata synthesises empty values in the same + // situation; GetResult later hard-asserts both are present. + kj::mv(rpc.httpMetadata).orDefault(HttpMetadata{}), + kj::mv(rpc.customMetadata).orDefault(jsg::Dict{}), kj::mv(rpc.range), + kj::mv(rpc.storageClass), kj::mv(rpc.ssecKeyMd5)); + addHeadResultSpanTags(js, traceContext, *result.get()); + return kj::mv(result); + } + return kj::none; + }); + }); +} + +jsg::Promise R2Bucket::deleteRpc(jsg::Lock& js, + kj::OneOf> keys, + const jsg::TypeHandler>& rpcPropType, + const jsg::TypeHandler>)>>& + fnType) { + return js.evalNow([&] { + auto& context = IoContext::current(); + TraceContext traceContext = context.makeUserTraceSpan("r2_delete"_kjc); + + traceContext.setTag("cloudflare.binding.type"_kjc, "r2"_kjc); + KJ_IF_SOME(b, this->bindingName()) { + traceContext.setTag("cloudflare.binding.name"_kjc, b); + } + traceContext.setTag("cloudflare.r2.operation"_kjc, "DeleteObject"_kjc); + KJ_IF_SOME(b, this->bucketName()) { + traceContext.setTag("cloudflare.r2.bucket"_kjc, b); + } + KJ_SWITCH_ONEOF(keys) { + KJ_CASE_ONEOF(ks, kj::Array) { + traceContext.setTag("cloudflare.r2.request.keys"_kjc, kj::str(ks)); + } + KJ_CASE_ONEOF(k, kj::String) { + traceContext.setTag("cloudflare.r2.request.keys"_kjc, kj::str(k)); + } + } + + auto rpcProp = getRpcMethod(js, "delete"_kj); + auto fn = JSG_REQUIRE_NONNULL(fnType.tryUnwrap(js, rpcPropType.wrap(js, kj::mv(rpcProp))), + Error, "R2 binding entrypoint's delete method is not callable"); + + // The result is discarded, matching delete_: a missing key is success, and per-key failures in + // a batch delete are reported in a body the binding has never read. + return normalizeRpcPromise(js, fn(js, kj::mv(keys))) + .then(js, [traceContext = kj::mv(traceContext)](jsg::Lock&, jsg::Value) mutable {}); + }); +} + R2Bucket::FeatureFlags::FeatureFlags(CompatibilityFlags::Reader featureFlags) : listHonorsIncludes(featureFlags.getR2ListHonorIncludeFields()) {} diff --git a/src/workerd/api/r2-bucket.h b/src/workerd/api/r2-bucket.h index a57770965f9..24733bdcc1b 100644 --- a/src/workerd/api/r2-bucket.h +++ b/src/workerd/api/r2-bucket.h @@ -7,7 +7,9 @@ #include "r2-rpc.h" #include +#include #include +#include namespace workerd::api { class Headers; @@ -242,6 +244,56 @@ class R2Bucket: public jsg::Object { JSG_STRUCT_TS_OVERRIDE(R2MultipartOptions); }; + // Object metadata as it crosses the JSRPC boundary, mirroring the shape the R2 + // gateway worker returns. Distinct from `HeadResult`, which is a resource type + // carrying methods and lazy accessors that RPC cannot serialize; these plain + // structs are unwrapped from the RPC result and used to build one. + // + // Not part of the public API: these are internal to the JSRPC transport and are + // never handed to user code, so they carry no TS overrides. + struct ChecksumsRpc { + jsg::Optional> md5; + jsg::Optional> sha1; + jsg::Optional> sha256; + jsg::Optional> sha384; + jsg::Optional> sha512; + + JSG_STRUCT(md5, sha1, sha256, sha384, sha512); + }; + + // Field names match the gateway's R2ObjectRpc, not HeadResult's members: the + // key arrives as `key` where HeadResult stores it as `name`. + // + // `kj::Maybe` rather than `jsg::Optional` throughout, because jsg::Optional + // accepts `undefined` but not `null`, and only kj::Maybe tolerates both. The + // gateway omits absent fields today, but that is an unenforced cross-repo + // invariant and a null would otherwise be a hard unwrap failure. + struct HeadResultRpc { + kj::String key; + kj::String version; + double size; + kj::String etag; + kj::Date uploaded; + kj::String storageClass; + ChecksumsRpc checksums; + kj::Maybe httpMetadata; + kj::Maybe> customMetadata; + kj::Maybe range; + kj::Maybe ssecKeyMd5; + + JSG_STRUCT(key, + version, + size, + etag, + uploaded, + storageClass, + checksums, + httpMetadata, + customMetadata, + range, + ssecKeyMd5); + }; + class HeadResult: public jsg::Object { public: HeadResult(kj::String name, @@ -490,18 +542,56 @@ class R2Bucket: public jsg::Object { jsg::Promise delete_(jsg::Lock& js, kj::OneOf> keys, const jsg::TypeHandler>& errorType); + + // JSRPC equivalents of the above, selected by JSG_RESOURCE_TYPE when the + // R2_BINDINGS_JSRPC autogate and the r2_bindings_jsrpc compatibility flag are + // both on. They dispatch to the gateway's R2BindingEntrypoint instead of + // synthesising an HTTP request, then rebuild the public result types from the + // plain data JSRPC delivers. + // + // These keep ordinary typed signatures rather than taking a raw + // v8::FunctionCallbackInfo the way KvNamespace::deleteBulk does. A raw-args + // passthrough cannot work here: it returns the JsRpcPromise directly, so the + // caller receives the gateway's plain data and R2Object's methods and sync + // accessors are gone. Reconstruction needs the resolved value, which needs a + // real promise, which needs a typed return, which needs an injected + // TypeHandler -- and TypeHandlers are only injected into typed signatures. + jsg::Promise>> headRpc(jsg::Lock& js, + kj::String key, + const jsg::TypeHandler>& rpcPropType, + const jsg::TypeHandler>& fnType, + const jsg::TypeHandler>& resultType); + jsg::Promise deleteRpc(jsg::Lock& js, + kj::OneOf> keys, + const jsg::TypeHandler>& rpcPropType, + const jsg::TypeHandler< + jsg::Function>)>>& fnType); jsg::Promise list(jsg::Lock& js, jsg::Optional options, const jsg::TypeHandler>& errorType, CompatibilityFlags::Reader flags); JSG_RESOURCE_TYPE(R2Bucket, CompatibilityFlags::Reader flags) { - JSG_METHOD(head); + // Two gates, and they do different jobs. The autogate is the fleet-wide kill switch, flipped + // per metal via Release Manager; it cannot distinguish accounts. The compatibility flag is what + // restricts the new transport to allowlisted workers, because it is marked $experimental and + // EWC decides who may opt in. Neither alone is sufficient. + // + // Only head and delete are migrated so far; the rest stay on the HTTP transport, including all + // of R2MultipartUpload, whose methods would additionally need the upload's key and uploadId + // threaded into the call. + if (util::Autogate::isEnabled(util::AutogateKey::R2_BINDINGS_JSRPC) && + flags.getR2BindingsJsrpc()) { + JSG_METHOD_NAMED(head, headRpc); + JSG_METHOD_NAMED(delete, deleteRpc); + } else { + JSG_METHOD(head); + JSG_METHOD_NAMED(delete, delete_); + } JSG_METHOD(get); JSG_METHOD(put); JSG_METHOD(createMultipartUpload); JSG_METHOD(resumeMultipartUpload); - JSG_METHOD_NAMED(delete, delete_); JSG_METHOD(list); JSG_TS_ROOT(); @@ -600,6 +690,11 @@ class R2Bucket: public jsg::Object { kj::Own getHttpClient(IoContext& context, TraceContext& traceContext); + // Look up a method on the gateway's entrypoint over this binding's subrequest + // channel. Which entrypoint that resolves to is decided by the channel's + // configuration, not here -- a JSRPC call carries no entrypoint name. + jsg::Ref getRpcMethod(jsg::Lock& js, kj::StringPtr methodName); + friend class R2MultipartUpload; }; diff --git a/src/workerd/api/r2.h b/src/workerd/api/r2.h index eb104bbef50..e36ee6b5052 100644 --- a/src/workerd/api/r2.h +++ b/src/workerd/api/r2.h @@ -17,6 +17,7 @@ namespace workerd::api::public_beta { api::public_beta::R2Bucket::Checksums, api::public_beta::R2Bucket::StringChecksums, \ api::public_beta::R2Bucket::HttpMetadata, api::public_beta::R2Bucket::ListOptions, \ api::public_beta::R2Bucket::ListResult, \ - api::public_beta::R2MultipartUpload::UploadPartOptions + api::public_beta::R2MultipartUpload::UploadPartOptions, \ + api::public_beta::R2Bucket::ChecksumsRpc, api::public_beta::R2Bucket::HeadResultRpc // The list of r2 types that are added to worker.c++'s JSG_DECLARE_ISOLATE_TYPE } // namespace workerd::api::public_beta diff --git a/src/workerd/api/tests/BUILD.bazel b/src/workerd/api/tests/BUILD.bazel index 05116d77b3d..affe980332f 100644 --- a/src/workerd/api/tests/BUILD.bazel +++ b/src/workerd/api/tests/BUILD.bazel @@ -344,6 +344,12 @@ wd_test( ], ) +wd_test( + src = "r2-jsrpc-test.wd-test", + args = ["--experimental"], + data = ["r2-jsrpc-test.js"], +) + wd_test( src = "r2-test.wd-test", args = ["--experimental"], diff --git a/src/workerd/api/tests/r2-jsrpc-test.js b/src/workerd/api/tests/r2-jsrpc-test.js new file mode 100644 index 00000000000..12cef2c6c9c --- /dev/null +++ b/src/workerd/api/tests/r2-jsrpc-test.js @@ -0,0 +1,186 @@ +// Copyright (c) 2017-2022 Cloudflare, Inc. +// Licensed under the Apache 2.0 license found in the LICENSE file or at: +// https://opensource.org/licenses/Apache-2.0 + +import assert from 'node:assert'; +import { WorkerEntrypoint } from 'cloudflare:workers'; + +// Stands in for the gateway's R2BindingEntrypoint. It returns the values the real gateway returns +// after its own parsing -- plain data in the binding's vocabulary, with a real Date and real +// ArrayBuffers -- rather than the HTTP wire shape. The C++ side rebuilds R2Object from this. +// +// Written out literally rather than generated, because the RPC contract between the two repos is +// the thing under test; a shared builder would let both sides drift together. + +const uploadedMs = 1700000000000; + +function bytes(...values) { + return new Uint8Array(values).buffer; +} + +function baseObject(overrides = {}) { + return { + key: 'some/key', + version: '00000000-0000-0000-0000-000000000001', + size: 1024, + etag: 'abc123', + uploaded: new Date(uploadedMs), + storageClass: 'Standard', + checksums: {}, + ...overrides, + }; +} + +export class R2BindingEntrypoint extends WorkerEntrypoint { + async head(key) { + switch (key) { + case 'missing': + // Object not found is null, not an error. + return null; + case 'checksums': + return baseObject({ + checksums: { md5: bytes(0xff, 0x00), sha1: bytes(0x01, 0x02) }, + }); + case 'metadata': + return baseObject({ + httpMetadata: { + contentType: 'text/plain', + cacheExpiry: new Date(uploadedMs + 1000), + }, + customMetadata: { colour: 'blue' }, + }); + case 'ranged': + return baseObject({ range: { offset: 10, length: 20 } }); + case 'ssec': + return baseObject({ ssecKeyMd5: 'deadbeef' }); + case 'boom': + // The gateway formats errors as `: ()` and throws a plain Error, + // reproducing what R2Result::throwIfError produces on the HTTP path. + throw new Error('head: no such bucket (10006)'); + default: + return baseObject(); + } + } + + async delete(keys) { + if (keys === 'boom') { + throw new Error('delete: bad keys (10021)'); + } + // Record what arrived so the test can assert the union survived marshalling. + globalThis.lastDeleteArg = keys; + } +} + +export const test = { + async test(ctrl, env, ctx) { + // Dispatch reaches the named entrypoint at all. If the binding targeted the default export, or + // either gate were off, this would throw "does not implement the method". + { + const obj = await env.BUCKET.head('plain'); + assert.strictEqual(obj.key, 'some/key'); + assert.strictEqual(obj.version, '00000000-0000-0000-0000-000000000001'); + assert.strictEqual(obj.size, 1024); + assert.strictEqual(obj.etag, 'abc123'); + assert.strictEqual(obj.storageClass, 'Standard'); + assert.strictEqual(obj.uploaded.getTime(), uploadedMs); + } + + // The result is a real R2Object, not the plain data the gateway sent. This is what a raw + // passthrough would lose. + { + const obj = await env.BUCKET.head('plain'); + assert.strictEqual(typeof obj.writeHttpMetadata, 'function'); + assert.strictEqual(typeof obj.checksums.toJSON, 'function'); + assert.strictEqual(obj.httpEtag, '"abc123"'); + } + + // Absent metadata becomes empty rather than undefined, matching the HTTP path. GetResult later + // hard-asserts both are present, so this is load-bearing beyond head(). + { + const obj = await env.BUCKET.head('plain'); + assert.deepStrictEqual(obj.httpMetadata, {}); + assert.deepStrictEqual(obj.customMetadata, {}); + } + + { + const obj = await env.BUCKET.head('missing'); + assert.strictEqual(obj, null); + } + + { + const obj = await env.BUCKET.head('checksums'); + assert.deepStrictEqual( + new Uint8Array(obj.checksums.md5), + new Uint8Array([0xff, 0x00]) + ); + assert.deepStrictEqual( + new Uint8Array(obj.checksums.sha1), + new Uint8Array([0x01, 0x02]) + ); + assert.strictEqual(obj.checksums.sha256, undefined); + assert.deepStrictEqual(obj.checksums.toJSON(), { + md5: 'ff00', + sha1: '0102', + }); + } + + { + const obj = await env.BUCKET.head('metadata'); + assert.strictEqual(obj.httpMetadata.contentType, 'text/plain'); + assert.strictEqual( + obj.httpMetadata.cacheExpiry.getTime(), + uploadedMs + 1000 + ); + assert.deepStrictEqual(obj.customMetadata, { colour: 'blue' }); + + const headers = new Headers(); + obj.writeHttpMetadata(headers); + assert.strictEqual(headers.get('content-type'), 'text/plain'); + } + + { + const obj = await env.BUCKET.head('ranged'); + assert.strictEqual(obj.range.offset, 10); + assert.strictEqual(obj.range.length, 20); + } + + { + const obj = await env.BUCKET.head('ssec'); + assert.strictEqual(obj.ssecKeyMd5, 'deadbeef'); + } + + // Errors cross RPC as a plain Error with the message already formatted. The v4 code is part of + // the text, not a property -- R2Result::throwIfError's structured R2Error throw is compiled out + // under `#if 0`, so pinning `.code === undefined` guards against silently changing that. + { + await assert.rejects(env.BUCKET.head('boom'), (err) => { + assert.strictEqual(err.message, 'head: no such bucket (10006)'); + assert.strictEqual(err.code, undefined); + return true; + }); + } + + // delete resolves to undefined and forwards the string/array union unchanged. + { + assert.strictEqual(await env.BUCKET.delete('one/key'), undefined); + assert.strictEqual(globalThis.lastDeleteArg, 'one/key'); + + await env.BUCKET.delete(['a', 'b', 'c']); + assert.deepStrictEqual(globalThis.lastDeleteArg, ['a', 'b', 'c']); + } + + { + await assert.rejects(env.BUCKET.delete('boom'), (err) => { + assert.strictEqual(err.message, 'delete: bad keys (10021)'); + return true; + }); + } + + // Argument coercion still happens client-side, because the C++ method keeps a typed kj::String + // parameter. A raw FunctionCallbackInfo passthrough would have sent the number through. + { + const obj = await env.BUCKET.head(12345); + assert.strictEqual(obj.key, 'some/key'); + } + }, +}; diff --git a/src/workerd/api/tests/r2-jsrpc-test.wd-test b/src/workerd/api/tests/r2-jsrpc-test.wd-test new file mode 100644 index 00000000000..42a12a65237 --- /dev/null +++ b/src/workerd/api/tests/r2-jsrpc-test.wd-test @@ -0,0 +1,31 @@ +using Workerd = import "/workerd/workerd.capnp"; + +# The JSRPC transport for R2 bindings. r2-test.wd-test covers the HTTP transport and stays as it +# is -- both paths ship simultaneously, selected by the gate below, so both need coverage. +# +# Reaching the JSRPC method table needs three things to line up: +# - the R2_BINDINGS_JSRPC autogate (the fleet-wide kill switch) +# - the r2_bindings_jsrpc compatibility flag, which is $experimental and so also needs +# "experimental" in the flag list here +# - a binding that targets the named entrypoint the gateway exposes +# +# The `entrypoint` below is the point of this file. A bare-string designator would target the +# default export and would not exercise named-entrypoint dispatch, which is the production shape. +const unitTests :Workerd.Config = ( + services = [ + ( name = "r2-jsrpc-test", + worker = ( + modules = [ + ( name = "worker", esModule = embed "r2-jsrpc-test.js" ) + ], + bindings = [ + ( name = "BUCKET", + r2Bucket = (name = "r2-jsrpc-test", entrypoint = "R2BindingEntrypoint") ), + ], + compatibilityFlags = ["experimental", "nodejs_compat", "r2_bindings_jsrpc", + "disable_fast_jsg_struct"], + ) + ), + ], + autogates = ["workerd-autogate-r2-bindings-jsrpc"], +); diff --git a/src/workerd/io/compatibility-date.capnp b/src/workerd/io/compatibility-date.capnp index 1ac5aa6220e..ffe95850a57 100644 --- a/src/workerd/io/compatibility-date.capnp +++ b/src/workerd/io/compatibility-date.capnp @@ -1651,4 +1651,14 @@ struct CompatibilityFlags @0x8f8c1b68151b6cef { # RangeError (JS API) or trap (wasm opcode). # WARNING: Do not remove the `$experimental` marker before # the v8 change becomes part of chrome's default config. + + r2BindingsJsrpc @187 :Bool + $compatEnableFlag("r2_bindings_jsrpc") + $experimental; + # When enabled, R2 bindings dispatch to the R2 gateway worker over JSRPC + # instead of synthesising an HTTP request and calling `fetch`. Without this + # flag, R2 bindings continue to use the HTTP transport. + # + # The JSRPC path is additionally gated on the R2_BINDINGS_JSRPC autogate, which + # is the fleet-wide kill switch; this flag controls which workers may opt in. } diff --git a/src/workerd/util/autogate.h b/src/workerd/util/autogate.h index 9f8b3eb5693..a114b693742 100644 --- a/src/workerd/util/autogate.h +++ b/src/workerd/util/autogate.h @@ -106,7 +106,11 @@ namespace workerd::util { /* Allow a Socket to be transferred over JS RPC. When disabled, serializing a Socket fails as \ though the type were not serializable at all, and an incoming transferred Socket is \ rejected. */ \ - V(SOCKET_RPC_TRANSFER) + V(SOCKET_RPC_TRANSFER) \ + /* Fleet-wide kill switch for the R2 bindings JSRPC transport. The JSRPC method table is only \ + selected when this is enabled AND the worker carries the r2_bindings_jsrpc compatibility \ + flag, which is what restricts it to allowlisted accounts. */ \ + V(R2_BINDINGS_JSRPC) // clang-format on // --------------------------------------------------------------------------------------