Skip to content
Closed
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
77 changes: 77 additions & 0 deletions src/workerd/api/jsrpc-call-plan-test.c++
Original file line number Diff line number Diff line change
Expand Up @@ -704,6 +704,22 @@ constexpr kj::StringPtr ACTOR_SOURCE = R"JS(
}
)JS"_kj;

constexpr kj::StringPtr LATE_REJECTION_ACTOR_SOURCE = R"JS(
import { DurableObject } from "cloudflare:workers";
globalThis.events = "";
globalThis.rejectPending = undefined;
export default class extends DurableObject {
startPending() {
globalThis.events += "startPending;";
return new Promise((resolve, reject) => { globalThis.rejectPending = reject; });
}
rejectPending() {
globalThis.events += "rejectPending;";
globalThis.rejectPending(new Error("late rejection"));
}
}
)JS"_kj;

struct StartedSession {
rpc::JsRpcTarget::Client cap;
kj::Promise<WorkerInterface::CustomEvent::Result> session;
Expand Down Expand Up @@ -787,7 +803,12 @@ class RetryClaimObserver final: public RequestObserver {
}
}

void reportFailure(const kj::Exception&, FailureSource) override {
++failureCount;
}

uint claimCount = 0;
uint failureCount = 0;
IsRetryableHandler retryableAtClaim = IsRetryableHandler::NO;
kj::String jsEventsAtClaim;
kj::Maybe<kj::Exception> rejection;
Expand All @@ -803,6 +824,17 @@ TestFixture::SetupParams retryClaimActorParams(RetryClaimObserver& observer) {
};
}

TestFixture::SetupParams observedActorParams(
RetryClaimObserver& observer, kj::StringPtr source, kj::StringPtr actorId) {
return {
.mainModuleSource = source,
.actorId = Worker::Actor::Id(kj::str(actorId)),
.actorClassName = "default"_kj,
.requestObserverFactory = kj::Function<kj::Own<RequestObserver>()>(
[&observer]() -> kj::Own<RequestObserver> { return kj::addRef(observer); }),
};
}

void call(rpc::JsRpcTarget::Client& cap, kj::StringPtr name, kj::WaitScope& waitScope) {
auto request = cap.callRequest();
request.setMethodName(name);
Expand Down Expand Up @@ -837,6 +869,51 @@ KJ_TEST("a JSRPC session without calls does not claim") {
session.wait(fixture.getWaitScope());

KJ_EXPECT(observer->claimCount == 0);
KJ_EXPECT(observer->failureCount == 0);
Comment thread
danlapid marked this conversation as resolved.
}

KJ_TEST("a thrown JSRPC method reports its request failure exactly once") {
auto observer = kj::refcounted<RetryClaimObserver>();
TestFixture fixture(observedActorParams(*observer, ACTOR_SOURCE, "jsrpc-failure-test"_kj));
auto entrypoint = fixture.makeWorkerEntrypoint();
auto [cap, session] = startSession(*entrypoint);

expectCallFailure(cap, "fail", fixture.getWaitScope());
KJ_EXPECT(observer->failureCount == 1);

cap = nullptr;
session.wait(fixture.getWaitScope());
KJ_EXPECT(observer->failureCount == 1);
}

KJ_TEST("a late JSRPC rejection after session cancellation is not reported") {
auto observer = kj::refcounted<RetryClaimObserver>();
TestFixture fixture(
observedActorParams(*observer, LATE_REJECTION_ACTOR_SOURCE, "jsrpc-late-rejection-test"_kj));

{
auto entrypoint = fixture.makeWorkerEntrypoint();
auto [cap, session] = startSession(*entrypoint);
auto request = cap.callRequest();
request.setMethodName("startPending");
auto pendingCall = request.send();

// Let the method save its rejector before dropping every handle that keeps the call live.
fixture.getWaitScope().poll();
KJ_EXPECT(getJsEvents(fixture) == "startPending;");
}

auto entrypoint = fixture.makeWorkerEntrypoint();
auto [cap, session] = startSession(*entrypoint);
call(cap, "rejectPending", fixture.getWaitScope());
fixture.getWaitScope().poll();

KJ_EXPECT(getJsEvents(fixture) == "startPending;rejectPending;");
KJ_EXPECT(observer->failureCount == 0);

cap = nullptr;
session.wait(fixture.getWaitScope());
KJ_EXPECT(observer->failureCount == 0);
}

KJ_TEST("calls on a stub returned by the top-level JSRPC call do not claim") {
Expand Down
10 changes: 10 additions & 0 deletions src/workerd/api/tests/BUILD.bazel
Original file line number Diff line number Diff line change
Expand Up @@ -247,6 +247,16 @@ wd_test(
tags = ["no-coverage"],
)

wd_test(
src = "jsrpc-error-trace-test.wd-test",
args = ["--experimental"],
data = [
"jsrpc-error-trace-test.js",
"jsrpc-error-trace-test-tail.js",
],
tags = ["no-coverage"],
)

# Regression test: jsrpc onset.info emitted into a streaming tail worker must remain
# serializable when forwarded over service-binding RPC, just like fetch onset.info.
wd_test(
Expand Down
127 changes: 127 additions & 0 deletions src/workerd/api/tests/jsrpc-error-trace-test-tail.js
Original file line number Diff line number Diff line change
@@ -0,0 +1,127 @@
// Copyright (c) 2026 Cloudflare, Inc.
// Licensed under the Apache 2.0 license found in the LICENSE file or at:
// https://opensource.org/licenses/Apache-2.0

import * as assert from 'node:assert';

const invocations = [];
const HUNG_REQUEST_MESSAGE =
"The Workers runtime canceled this request because it detected that your Worker's code had hung and would never generate a response. Refer to: https://developers.cloudflare.com/workers/observability/errors/";

export default {
tailStream(onset) {
if (
onset.event.info?.type !== 'jsrpc' ||
onset.event.entrypoint !== 'ThrowingService'
) {
return () => {};
}

const invocation = {
methods: [],
logs: [],
exceptions: [],
outcomes: [],
};
invocations.push(invocation);

return (event) => {
const item = event.event;
if (item.type === 'attributes') {
const method = item.info.find(({ name }) => name === 'jsrpc.method');
const targetKind = item.info.find(
({ name }) => name === 'jsrpc.target_kind'
);
const operation = item.info.find(
({ name }) => name === 'jsrpc.operation'
);
if (
method?.value &&
targetKind?.value === 'entrypoint' &&
operation?.value === 'call'
) {
invocation.methods.push(method.value);
}
} else if (item.type === 'log') {
invocation.logs.push(item.message);
} else if (item.type === 'exception') {
invocation.exceptions.push({ name: item.name, message: item.message });
} else if (item.type === 'outcome') {
invocation.outcomes.push(item.outcome);
}
};
},
};

function findInvocation(method) {
return invocations.find((invocation) => invocation.methods.includes(method));
}

function hasException(invocation, message) {
return invocation?.exceptions.some(
(exception) => exception.message === message
);
}

function foundExpectedEvents(throwing, asyncThrowing) {
return (
hasException(throwing, 'intentional JSRPC failure') &&
throwing?.outcomes.includes('exception') &&
hasException(asyncThrowing, 'intentional async JSRPC failure') &&
asyncThrowing?.outcomes.includes('exception')
);
}

export const test = {
async test() {
const deadline = Date.now() + 5000;
let throwing = findInvocation('throwError');
let asyncThrowing = findInvocation('throwAsyncError');

while (
!foundExpectedEvents(throwing, asyncThrowing) &&
Date.now() < deadline
) {
await scheduler.wait(10);
throwing = findInvocation('throwError');
asyncThrowing = findInvocation('throwAsyncError');
}

assert.ok(throwing, 'Could not find throwError JSRPC invocation');
assert.ok(asyncThrowing, 'Could not find throwAsyncError JSRPC invocation');
assert.strictEqual(
invocations.length,
2,
'Expected exactly two callee invocations'
);

assert.deepStrictEqual(throwing.methods, ['throwError']);
const intentionalFailure = {
name: 'Error',
message: 'intentional JSRPC failure',
};
// Invocation cleanup may also emit its cancellation exception, depending on scheduling.
const expectedThrowingExceptions = [
intentionalFailure,
...(throwing.exceptions.length === 1
? []
: [
{
name: 'Error',
message: HUNG_REQUEST_MESSAGE,

Copy link
Copy Markdown
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

This is the bug we want to get rid of and this asserts that it's expected behaviour (...sometimes).

I'm not at all convinced this is directionally the right fix. I think we need to add a Rejected state in worker-rpc.h: kj::OneOf<Pending, Resolved, Disposed> state -> kj::OneOf<Pending, Resolved, Disposed, Rejected> state where Rejected drops the PendingPipeline reference that Pending is currently holding onto even after receiving a rejection. That is what's keeping the RPC session and the callee's request context alive leading to the hang detector kicking in on the callee.

},
]),
];
assert.deepStrictEqual(throwing.exceptions, expectedThrowingExceptions);
assert.deepStrictEqual(throwing.outcomes, ['exception']);

assert.deepStrictEqual(asyncThrowing.methods, ['throwAsyncError']);
assert.deepStrictEqual(asyncThrowing.exceptions, [
{
name: 'Error',
message: 'intentional async JSRPC failure',
},
]);
assert.deepStrictEqual(asyncThrowing.outcomes, ['exception']);
},
};
36 changes: 36 additions & 0 deletions src/workerd/api/tests/jsrpc-error-trace-test.js
Original file line number Diff line number Diff line change
@@ -0,0 +1,36 @@
// Copyright (c) 2026 Cloudflare, Inc.
// Licensed under the Apache 2.0 license found in the LICENSE file or at:
// https://opensource.org/licenses/Apache-2.0

import { strictEqual } from 'node:assert';
import { WorkerEntrypoint } from 'cloudflare:workers';

export class ThrowingService extends WorkerEntrypoint {
throwError() {
console.log('callee throwError called');
throw new Error('intentional JSRPC failure');
}

async throwAsyncError() {
await scheduler.wait(1);
throw new Error('intentional async JSRPC failure');
}
}

export default {
async test(_controller, env, _ctx) {
try {
await env.ThrowingService.throwError();
throw new Error('Expected throwError() to reject');
} catch (error) {
strictEqual(error.message, 'intentional JSRPC failure');
}

try {
await env.ThrowingService.throwAsyncError();
throw new Error('Expected throwAsyncError() to reject');
} catch (error) {
strictEqual(error.message, 'intentional async JSRPC failure');
}
},
};
28 changes: 28 additions & 0 deletions src/workerd/api/tests/jsrpc-error-trace-test.wd-test
Original file line number Diff line number Diff line change
@@ -0,0 +1,28 @@
using Workerd = import "/workerd/workerd.capnp";

# Test JSRPC callee-side tracing for delivered method failures.

const unitTests :Workerd.Config = (
services = [
(name = "jsrpc-error-trace-test", worker = .mainWorker),
(name = "tail", worker = .tailWorker),
],
);

const mainWorker :Workerd.Worker = (
modules = [
(name = "worker", esModule = embed "jsrpc-error-trace-test.js")
],
compatibilityFlags = ["nodejs_compat", "experimental", "rpc", "enhanced_error_serialization"],
bindings = [
(name = "ThrowingService", service = (name = "jsrpc-error-trace-test", entrypoint = "ThrowingService")),
],
streamingTails = ["tail"],
);

const tailWorker :Workerd.Worker = (
modules = [
(name = "worker", esModule = embed "jsrpc-error-trace-test-tail.js")
],
compatibilityFlags = ["experimental", "nodejs_compat"],
);
Loading
Loading