Skip to content

Await sibling deliveries after a fan-out failure #1388

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Open
dahlia wants to merge 1 commit into
base: 2.0-maintenance
Choose a base branch
from
+326 −12
Open

Await sibling deliveries after a fan-out failure #1388

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
1 commit
Select commit
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
9 changes: 9 additions & 0 deletions CHANGES.md
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -10,6 +10,13 @@ To be released.

### @fedify/fedify

- Fixed `Context.sendActivity()` and `InboxContext.forwardActivity()` giving
up on the remaining deliveries when one recipient failed. The immediate
delivery paths now wait for every delivery they have already started before
propagating the error, so the remaining recipients are still reached on
runtimes such as Cloudflare Workers, which may terminate an isolate once
the error response is returned. This applies both to `immediate: true` and
to federations without an outbox queue. [[#1380], [#1388]]
- Fixed Linked Data Signatures being emitted with a JSON-LD context that no
longer resolves. The attached `signature` object now uses
`https://w3id.org/security/v1` instead of
Expand Down Expand Up @@ -46,7 +53,9 @@ To be released.
[#664]: https://github.com/fedify-dev/fedify/issues/664
[#1339]: https://github.com/fedify-dev/fedify/issues/1339
[#1347]: https://github.com/fedify-dev/fedify/pull/1347
[#1380]: https://github.com/fedify-dev/fedify/issues/1380
[#1381]: https://github.com/fedify-dev/fedify/issues/1381
[#1388]: https://github.com/fedify-dev/fedify/pull/1388
[#1390]: https://github.com/fedify-dev/fedify/pull/1390

### @fedify/vocab
Expand Down
12 changes: 12 additions & 0 deletions changes.d/fedify/await-sibling-deliveries.md
View file
Open in desktop
Original file line number Diff line number Diff line change
@@ -0,0 +1,12 @@
---
links:
'#1380': https://github.com/fedify-dev/fedify/issues/1380
'#1388': https://github.com/fedify-dev/fedify/pull/1388
---
- Fixed `Context.sendActivity()` and `InboxContext.forwardActivity()` giving
up on the remaining deliveries when one recipient failed. The immediate
delivery paths now wait for every delivery they have already started before
propagating the error, so the remaining recipients are still reached on
runtimes such as Cloudflare Workers, which may terminate an isolate once
the error response is returned. This applies both to `immediate: true` and
to federations without an outbox queue. [[#1380], [#1388]]
270 changes: 270 additions & 0 deletions packages/fedify/src/federation/middleware.test.ts
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -65,6 +65,7 @@ import {
import type { MessageQueue } from "./mq.ts";
import type { InboxMessage, Message, OutboxMessage } from "./queue.ts";
import { RouterError } from "./router.ts";
import { SendActivityError } from "./send.ts";

test("createFederation()", async (t) => {
const kv = new MemoryKvStore();
Expand Down Expand Up @@ -7275,6 +7276,275 @@ test("ContextImpl.sendActivity() honors the private-address policy", async (t) =
}
});

interface DeferredSignal {
readonly promise: Promise<void>;
readonly resolve: () => void;
}

function deferredSignal(): DeferredSignal {
let resolve!: () => void;
const promise = new Promise<void>((r) => resolve = r);
return { promise, resolve };
}

test("ContextImpl.sendActivity() waits for sibling deliveries on failure", async () => {
const failingInbox = "http://127.0.0.1:3001/inbox/fail";
const slowInbox = "http://127.0.0.1:3002/inbox/slow";
const failingEntered = deferredSignal();
const slowEntered = deferredSignal();
const slowCompleted = deferredSignal();
let releaseSlow!: () => void;
const slowGate = new Promise<void>((resolve) => releaseSlow = resolve);
let slowStarted = false;
let slowSettled = false;

fetchMock.spyGlobal();
fetchMock.post(failingInbox, () => {
failingEntered.resolve();
return new Response(null, { status: 500 });
});
fetchMock.post(slowInbox, async () => {
slowStarted = true;
slowEntered.resolve();
await slowGate;
slowSettled = true;
slowCompleted.resolve();
return new Response(null, { status: 202 });
});

let observed: Promise<void> | undefined;
let released = false;
try {
const federation = createFederation<void>({
kv: new MemoryKvStore(),
allowPrivateAddress: true,
});
const ctx = federation.createContext(new URL("https://example.com/"));
const activity = new Create({
id: new URL("https://example.com/activity/fanout"),
actor: new URL("https://example.com/person"),
});
let outerSettled = false;
let outerError: unknown;
const sending = ctx.sendActivity(
[{ privateKey: ed25519PrivateKey, keyId: ed25519Multikey.id! }],
[
new Person({
id: new URL("https://example.com/recipients/failing"),
inbox: new URL(failingInbox),
}),
new Person({
id: new URL("https://example.com/recipients/slow"),
inbox: new URL(slowInbox),
}),
],
activity,
{ immediate: true },
);
observed = sending.then(
() => {
outerSettled = true;
},
(error) => {
outerSettled = true;
outerError = error;
},
);
await Promise.all([failingEntered.promise, slowEntered.promise]);
await new Promise((resolve) => setTimeout(resolve, 0));
assertFalse(
outerSettled,
"sendActivity() must wait for sibling deliveries before rejecting",
);
releaseSlow();
released = true;
await observed;
assert(slowSettled, "the delayed delivery should have settled");
assertInstanceOf(outerError, SendActivityError);
} finally {
if (!released) releaseSlow();
if (slowStarted) await slowCompleted.promise;
await observed;
fetchMock.hardReset();
}
});

test("InboxContextImpl.forwardActivity() waits for sibling deliveries on failure", async () => {
const failingInbox = "http://127.0.0.1:3003/inbox/fail";
const slowInbox = "http://127.0.0.1:3004/inbox/slow";
const failingEntered = deferredSignal();
const slowEntered = deferredSignal();
const slowCompleted = deferredSignal();
let releaseSlow!: () => void;
const slowGate = new Promise<void>((resolve) => releaseSlow = resolve);
let slowStarted = false;
let slowSettled = false;

fetchMock.spyGlobal();
fetchMock.post(failingInbox, () => {
failingEntered.resolve();
return new Response(null, { status: 500 });
});
fetchMock.post(slowInbox, async () => {
slowStarted = true;
slowEntered.resolve();
await slowGate;
slowSettled = true;
slowCompleted.resolve();
return new Response(null, { status: 202 });
});

let observed: Promise<void> | undefined;
let released = false;
try {
const federation = new FederationImpl<void>({
kv: new MemoryKvStore(),
allowPrivateAddress: true,
});
const activity = await signObject(
new vocab.Create({
id: new URL("https://example.com/activity/forward"),
actor: new URL("https://example.com/person2"),
}),
ed25519PrivateKey,
ed25519Multikey.id!,
{ contextLoader: mockDocumentLoader, documentLoader: mockDocumentLoader },
);
const ctx = new InboxContextImpl(
null,
await activity.toJsonLd({ contextLoader: mockDocumentLoader }),
activity.id?.href,
getTypeId(activity).href,
{
data: undefined,
federation,
url: new URL("https://example.com/"),
documentLoader: mockDocumentLoader,
contextLoader: mockDocumentLoader,
},
);
let outerSettled = false;
let outerError: unknown;
const sending = ctx.forwardActivity(
[{ privateKey: ed25519PrivateKey, keyId: ed25519Multikey.id! }],
[
{
id: new URL("https://example.com/recipients/failing"),
inboxId: new URL(failingInbox),
},
{
id: new URL("https://example.com/recipients/slow"),
inboxId: new URL(slowInbox),
},
],
{ immediate: true, skipIfUnsigned: true },
);
observed = sending.then(
() => {
outerSettled = true;
},
(error) => {
outerSettled = true;
outerError = error;
},
);
await Promise.all([failingEntered.promise, slowEntered.promise]);
await new Promise((resolve) => setTimeout(resolve, 0));
assertFalse(
outerSettled,
"forwardActivity() must wait for sibling deliveries before rejecting",
);
releaseSlow();
released = true;
await observed;
assert(slowSettled, "the delayed delivery should have settled");
assertInstanceOf(outerError, SendActivityError);
} finally {
if (!released) releaseSlow();
if (slowStarted) await slowCompleted.promise;
await observed;
fetchMock.hardReset();
}
});

test("FederationImpl.sendActivity() drains deliveries when fan-out setup fails", async () => {
const slowInbox = "http://127.0.0.1:3005/inbox/slow";
const slowEntered = deferredSignal();
const slowCompleted = deferredSignal();
let releaseSlow!: () => void;
const slowGate = new Promise<void>((resolve) => releaseSlow = resolve);
let slowStarted = false;
let slowSettled = false;

fetchMock.spyGlobal();
fetchMock.post(slowInbox, async () => {
slowStarted = true;
slowEntered.resolve();
await slowGate;
slowSettled = true;
slowCompleted.resolve();
return new Response(null, { status: 202 });
});

let observed: Promise<void> | undefined;
let released = false;
try {
const federation = new FederationImpl<void>({
kv: new MemoryKvStore(),
allowPrivateAddress: true,
});
const context = federation.createContext(new URL("https://example.com/"));
const activity = new Create({
id: new URL("https://example.com/activity/setup-failure"),
actor: new URL("https://example.com/person"),
});
let outerSettled = false;
let outerError: unknown;
const sending = federation.sendActivity(
[{ privateKey: ed25519PrivateKey, keyId: ed25519Multikey.id! }],
{
[slowInbox]: {
actorIds: ["https://example.com/recipients/slow"],
sharedInbox: false,
},
// Synthetic invalid input: the fan-out fails while preparing the
// second recipient, after the first delivery has already started.
"not a url": {
actorIds: ["https://example.com/recipients/invalid"],
sharedInbox: false,
},
},
activity,
{ context, immediate: true },
);
observed = sending.then(
() => {
outerSettled = true;
},
(error) => {
outerSettled = true;
outerError = error;
},
);
await slowEntered.promise;
await new Promise((resolve) => setTimeout(resolve, 0));
assertFalse(
outerSettled,
"sendActivity() must drain started deliveries when fan-out setup fails",
);
releaseSlow();
released = true;
await observed;
assert(slowSettled, "the started delivery should have settled");
assertInstanceOf(outerError, TypeError);
} finally {
if (!released) releaseSlow();
if (slowStarted) await slowCompleted.promise;
await observed;
fetchMock.hardReset();
}
});

test("Federation.fetch() bounds inbox bodies before dispatch", async () => {
let dispatched = false;
const json = JSON.stringify({
Expand Down
47 changes: 35 additions & 12 deletions packages/fedify/src/federation/middleware.ts
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -1473,9 +1473,9 @@ export class FederationImpl<TContextData>
);
}
const promises: Promise<void>[] = [];
for (const inbox in inboxes) {
promises.push(
sendActivity({
try {
for (const inbox in inboxes) {
const promise = sendActivity({
allowPrivateAddress: this.allowPrivateAddress,
keys,
activity: jsonLd,
Expand All @@ -1496,10 +1496,22 @@ export class FederationImpl<TContextData>
this.firstKnock,
),
tracerProvider: this.tracerProvider,
}),
);
});
// Mark the delivery as handled so an early rejection does not
// surface as an unhandled rejection while a later recipient is
// still being prepared. Promise.all() and Promise.allSettled()
// still observe the original rejection.
promise.catch(() => {});
promises.push(promise);
}
await Promise.all(promises);
} finally {
// Drain the deliveries that were already started before propagating
// the original error, so that a failing recipient cannot cancel its
// siblings when the runtime tears down the request context (e.g.,
// Cloudflare Workers).
await Promise.allSettled(promises);
}
await Promise.all(promises);
return;
}
logger.debug(
Expand Down Expand Up @@ -3432,9 +3444,9 @@ export class InboxContextImpl<TContextData> extends ContextImpl<TContextData>
);
}
const promises: Promise<void>[] = [];
for (const inbox in inboxes) {
promises.push(
sendActivity({
try {
for (const inbox in inboxes) {
const promise = sendActivity({
allowPrivateAddress: this.federation.allowPrivateAddress,
keys,
activity: this.activity,
Expand All @@ -3448,10 +3460,21 @@ export class InboxContextImpl<TContextData> extends ContextImpl<TContextData>
this.federation.kvPrefixes.httpMessageSignaturesSpec,
this.federation.firstKnock,
),
}),
);
});
// Mark the delivery as handled so an early rejection does not
// surface as an unhandled rejection if the loop fails before
// Promise.all() attaches its handlers.
promise.catch(() => {});
promises.push(promise);
}
await Promise.all(promises);
} finally {
// Drain the deliveries that were already started before propagating
// the original error, so that a failing recipient cannot cancel its
// siblings when the runtime tears down the request context (e.g.,
// Cloudflare Workers).
await Promise.allSettled(promises);
}
await Promise.all(promises);
return;
}
logger.debug(
Expand Down
Loading