Unverified Commit 5c16bd70 authored by Hong Minhee's avatar Hong Minhee
Browse files

Instrument `sendActivity()`

parent f753d606

docs/manual/opentelemetry.md

+1 −0
Original line number Diff line number Diff line
@@ -128,6 +128,7 @@ spans: 
| `activitypub.get_actor_handle`                      | Client      | Resolves the actor handle.                  |

| `activitypub.lookup_object`                         | Client      | Looks up the Activity Streams object.       |

| `activitypub.parse_object`                          | Internal    | Parses the Activity Streams object.         |

| `activitypub.send_activity`                         | Client      | Sends the ActivityPub activity.             |

| `http_signatures.sign`                              | Internal    | Signs the HTTP request.                     |

| `http_signatures.verify`                            | Internal    | Verifies the HTTP request signature.        |

| `ld_signatures.sign`                                | Internal    | Makes the Linked Data signature.            |

src/federation/handler.ts

+12 −1
Original line number Diff line number Diff line
@@ -10,6 +10,7 @@ import { detachSignature, verifyJsonLd } from "../sig/ld.ts"; 
import { doesActorOwnKey } from "../sig/owner.ts";

import { verifyObject } from "../sig/proof.ts";

import type { Recipient } from "../vocab/actor.ts";

import { getTypeId } from "../vocab/type.ts";

import {

  Activity,

  CryptographicKey,
@@ -411,6 +412,8 @@ export interface InboxHandlerParameters<TContextData> { 
  inboxContextFactory(

    recipient: string | null,

    activity: unknown,

    activityId: string | undefined,

    activityType: string,

  ): InboxContext<TContextData>;

  kv: KvStore;

  kvPrefixes: {
@@ -683,7 +686,15 @@ export async function handleInbox<TContextData>( 
    });

  }

  try {

    await listener(inboxContextFactory(recipient, json), activity);

    await listener(

      inboxContextFactory(

        recipient,

        json,

        activity.id?.href,

        getTypeId(activity).href,

      ),

      activity,

    );

  } catch (error) {

    try {

      await inboxErrorHandler?.(context, error as Error);

src/federation/middleware.test.ts

+39 −18
Original line number Diff line number Diff line
@@ -29,6 +29,7 @@ import { 
} from "../testing/keys.ts";

import { test } from "../testing/mod.ts";

import { lookupObject } from "../vocab/lookup.ts";

import { getTypeId } from "../vocab/type.ts";

import {

  Activity,

  Create,
@@ -1217,12 +1218,18 @@ test("InboxContextImpl.forwardActivity()", async (t) => { 
      "id": "https://example.com/activity",

      "actor": "https://example.com/person2",

    };

    const ctx = new InboxContextImpl(null, activity, {

    const ctx = new InboxContextImpl(

      null,

      activity,

      "https://example.com/activity",

      "https://www.w3.org/ns/activitystreams#Create",

      {

        data: undefined,

        federation,

        url: new URL("https://example.com/"),

        documentLoader: fetchDocumentLoader,

    });

      },

    );

    await ctx.forwardActivity(

      [{ privateKey: rsaPrivateKey2, keyId: rsaPublicKey2.id! }],

      {
@@ -1241,12 +1248,18 @@ test("InboxContextImpl.forwardActivity()", async (t) => { 
      "id": "https://example.com/activity",

      "actor": "https://example.com/person2",

    };

    const ctx = new InboxContextImpl(null, activity, {

    const ctx = new InboxContextImpl(

      null,

      activity,

      "https://example.com/activity",

      "https://www.w3.org/ns/activitystreams#Create",

      {

        data: undefined,

        federation,

        url: new URL("https://example.com/"),

        documentLoader: fetchDocumentLoader,

    });

      },

    );

    await assertRejects(() =>

      ctx.forwardActivity(

        [{ privateKey: rsaPrivateKey2, keyId: rsaPublicKey2.id! }],
@@ -1272,6 +1285,8 @@ test("InboxContextImpl.forwardActivity()", async (t) => { 
    const ctx = new InboxContextImpl(

      null,

      await activity.toJsonLd({ contextLoader: mockDocumentLoader }),

      activity.id?.href,

      getTypeId(activity).href,

      {

        data: undefined,

        federation,
@@ -1302,12 +1317,18 @@ test("InboxContextImpl.forwardActivity()", async (t) => { 
      rsaPublicKey3.id!,

      { contextLoader: mockDocumentLoader },

    );

    const ctx = new InboxContextImpl(null, activity, {

    const ctx = new InboxContextImpl(

      null,

      activity,

      "https://example.com/activity",

      "https://www.w3.org/ns/activitystreams#Create",

      {

        data: undefined,

        federation,

        url: new URL("https://example.com/"),

        documentLoader: fetchDocumentLoader,

    });

      },

    );

    await ctx.forwardActivity(

      [{ privateKey: rsaPrivateKey2, keyId: rsaPublicKey2.id! }],

      {

src/federation/middleware.ts

+31 −21
Original line number Diff line number Diff line
@@ -534,7 +534,9 @@ export class FederationImpl<TContextData> implements Federation<TContextData> { 
        keys,

        activity: message.activity,

        activityId: message.activityId,

        activityType: message.activityType,

        inbox: new URL(message.inbox),

        sharedInbox: message.sharedInbox,

        headers: new Headers(message.headers),

        tracerProvider: this.tracerProvider,

      });
@@ -647,7 +649,12 @@ export class FederationImpl<TContextData> implements Federation<TContextData> { 
    }

    try {

      await listener(

        context.toInboxContext(message.identifier, message.activity),

        context.toInboxContext(

          message.identifier,

          message.activity,

          activity.id?.href,

          getTypeId(activity).href,

        ),

        activity,

      );

    } catch (error) {
@@ -1941,12 +1948,14 @@ export class FederationImpl<TContextData> implements Federation<TContextData> { 
            keys,

            activity: jsonLd,

            activityId: activity.id?.href,

            activityType: getTypeId(activity).href,

            inbox: new URL(inbox),

            sharedInbox: inboxes[inbox].sharedInbox,

            headers: collectionSync == null ? undefined : new Headers({

              "Collection-Synchronization":

                await buildCollectionSynchronizationHeader(

                  collectionSync,

                  inboxes[inbox],

                  inboxes[inbox].actorIds,

                ),

            }),

            tracerProvider: this.tracerProvider,
@@ -1973,14 +1982,16 @@ export class FederationImpl<TContextData> implements Federation<TContextData> { 
        keys: keyJwkPairs,

        activity: jsonLd,

        activityId: activity.id?.href,

        activityType: getTypeId(activity).href,

        inbox,

        sharedInbox: inboxes[inbox].sharedInbox,

        started: new Date().toISOString(),

        attempt: 0,

        headers: collectionSync == null ? {} : {

          "Collection-Synchronization":

            await buildCollectionSynchronizationHeader(

              collectionSync,

              inboxes[inbox],

              inboxes[inbox].actorIds,

            ),

        },

      };
@@ -2301,8 +2312,10 @@ export class ContextImpl<TContextData> implements Context<TContextData> { 
  toInboxContext(

    recipient: string | null,

    activity: unknown,

    activityId: string | undefined,

    activityType: string,

  ): InboxContextImpl<TContextData> {

    return new InboxContextImpl(recipient, activity, {

    return new InboxContextImpl(recipient, activity, activityId, activityType, {

      url: this.url,

      federation: this.federation,

      data: this.data,
@@ -3028,15 +3041,21 @@ export class InboxContextImpl<TContextData> extends ContextImpl<TContextData> 
  implements InboxContext<TContextData> {

  readonly recipient: string | null;

  readonly activity: unknown;

  readonly activityId?: string;

  readonly activityType: string;



  constructor(

    recipient: string | null,

    activity: unknown,

    activityId: string | undefined,

    activityType: string,

    options: ContextOptions<TContextData>,

  ) {

    super(options);

    this.recipient = recipient;

    this.activity = activity;

    this.activityId = activityId;

    this.activityType = activityType;

  }



  forwardActivity(
@@ -3119,12 +3138,10 @@ export class InboxContextImpl<TContextData> extends ContextImpl<TContextData> 
    } else {

      keys = [forwarder];

    }

    let activityId: string | undefined = undefined;

    if (!hasSignature(this.activity)) {

      let hasProof: boolean;

      try {

        const activity = await Activity.fromJsonLd(this.activity, this);

        activityId = activity.id?.href;

        hasProof = await activity.getProof() != null;

      } catch {

        hasProof = false;
@@ -3138,17 +3155,6 @@ export class InboxContextImpl<TContextData> extends ContextImpl<TContextData> 
        );

      }

    }

    if (

      activityId == null && typeof this.activity === "object" &&

      this.activity != null

    ) {

      activityId =

        "@id" in this.activity && typeof this.activity["@id"] === "string"

          ? this.activity["@id"]

          : "id" in this.activity && typeof this.activity.id === "string"

          ? this.activity.id

          : undefined;

    }

    if (recipients === "followers") {

      if (identifier == null) {

        throw new Error(
@@ -3169,7 +3175,7 @@ export class InboxContextImpl<TContextData> extends ContextImpl<TContextData> 
    });

    logger.debug("Forwarding activity {activityId} to inboxes:\n{inboxes}", {

      inboxes: globalThis.Object.keys(inboxes),

      activityId,

      activityId: this.activityId,

      activity: this.activity,

    });

    if (options?.immediate || this.federation.outboxQueue == null) {
@@ -3190,8 +3196,10 @@ export class InboxContextImpl<TContextData> extends ContextImpl<TContextData> 
          sendActivity({

            keys,

            activity: this.activity,

            activityId: activityId,

            activityId: this.activityId,

            activityType: this.activityType,

            inbox: new URL(inbox),

            sharedInbox: inboxes[inbox].sharedInbox,

            tracerProvider: this.tracerProvider,

          }),

        );
@@ -3201,7 +3209,7 @@ export class InboxContextImpl<TContextData> extends ContextImpl<TContextData> 
    }

    logger.debug(

      "Enqueuing activity {activityId} to forward later.",

      { activityId, activity: this.activity },

      { activityId: this.activityId, activity: this.activity },

    );

    const keyJwkPairs: SenderKeyJwkPair[] = [];

    for (const { keyId, privateKey } of keys) {
@@ -3214,8 +3222,10 @@ export class InboxContextImpl<TContextData> extends ContextImpl<TContextData> 
        id: crypto.randomUUID(),

        keys: keyJwkPairs,

        activity: this.activity,

        activityId,

        activityId: this.activityId,

        activityType: this.activityType,

        inbox,

        sharedInbox: inboxes[inbox].sharedInbox,

        started: new Date().toISOString(),

        attempt: 0,

        headers: {},

src/federation/queue.ts

+2 −0
Original line number Diff line number Diff line
@@ -11,7 +11,9 @@ export interface OutboxMessage { 
  keys: SenderKeyJwkPair[];

  activity: unknown;

  activityId?: string;

  activityType: string;

  inbox: string;

  sharedInbox: boolean;

  started: string;

  attempt: number;

  headers: Record<string, string>;