536940436

Build Type

github

Committed by GitHub

Commit Message

Merge 24f0033ea into 1724ea31d

Pull Request Pull Request #76: Feature: Retry/Backoff

Run Details

419 of 638 branches covered (65.67%)

Branch coverage included in aggregate %.

27 of 28 new or added lines in 2 files covered. (96.43%)

28 existing lines in 2 files now uncovered.

857 of 1055 relevant lines covered (81.23%)

19.48 hits per line

Source File
Press 'n' to go to next uncovered line, 'b' for previous

82.93

/quirrel/src/api/worker/index.ts

import { decodeQueueDescriptor } from "../shared/queue-descriptor";
import { UsageMeter } from "../shared/usage-meter";
import fetch from "cross-fetch";
import { TokenRepo } from "../shared/token-repo";
import { sign } from "secure-webhooks";
import { Redis } from "ioredis";
import { Telemetrist } from "../shared/telemetrist";
import { createOwl } from "../shared/owl";
import type { Logger } from "../shared/logger";
import { IncidentForwarder } from "../shared/incident-forwarder";

interface ExecutionError {
  toString(): string;
  endpoint: string;
  tokenId: string;
  responseBody: string;
  responseStatus: number;
}

export function replaceLocalhostWithDockerHost(url: string): string {
  if (url.startsWith("http://localhost")) {
    return url.replace("http://localhost", "http://host.docker.internal");
  }

  if (url.startsWith("https://localhost")) {
    return url.replace("https://localhost", "https://host.docker.internal");
  }

  return url;
}

export interface QuirrelWorkerConfig {
  redisFactory: () => Redis;
  enableUsageMetering?: boolean;
  runningInDocker?: boolean;
  concurrency?: number;
  disableTelemetry?: boolean;
  logger?: Logger;
  incidentReceiver?: { endpoint: string; passphrase: string };
}

export async function createWorker({
  redisFactory,
  enableUsageMetering,
  runningInDocker,
  concurrency = 100,
  disableTelemetry,
  logger,
  incidentReceiver,
}: QuirrelWorkerConfig) {
  const incidentForwarder = incidentReceiver
    ? new IncidentForwarder(
        incidentReceiver.endpoint,
        incidentReceiver.passphrase
      )
    : undefined;

  const redisClient = redisFactory();
  const telemetrist = disableTelemetry
    ? undefined
    : new Telemetrist(runningInDocker ?? false);

  const tokenRepo = new TokenRepo(redisClient);

  let usageMeter: UsageMeter | undefined = undefined;
  if (enableUsageMetering) {
    usageMeter = new UsageMeter(redisClient);
  }

  const owl = createOwl(redisFactory);

  const worker = owl.createWorker(
    async (job, jobMeta) => {
      let { tokenId, endpoint } = decodeQueueDescriptor(job.queue);
      const body = job.payload;

      const executionDone = logger?.startingExecution({
        tokenId,
        endpoint,
        body,
        id: job.id,
      });

      const headers: Record<string, string> = {
        "Content-Type": "text/plain",
      };

      if (tokenId) {
        const token = await tokenRepo.getById(tokenId);
        if (token) {
          const signature = sign(body ?? "", token);
          headers["x-quirrel-signature"] = signature;
        }
      }

      if (runningInDocker) {
        endpoint = replaceLocalhostWithDockerHost(endpoint);
      }

      const [response] = await Promise.all([
        fetch(endpoint, {
          method: "POST",
          body,
          headers,
        }),
        usageMeter?.record(tokenId),
      ]);

      if (response.status >= 200 && response.status < 300) {
        executionDone?.();

        telemetrist?.dispatch("dispatch_job");
      } else {
        const responseBody = await response.text();

        logger?.executionErrored(
          { endpoint, tokenId, body: job.payload, id: job.id },
          responseBody
        );

        if (response.status === 404) {
          jobMeta.dontReschedule();
        }

        const error: ExecutionError = {
          tokenId,
          endpoint,
          responseBody,
          responseStatus: response.status,
          toString() {
            return responseBody;
          },
        };

        throw error;
      }
    },
    async (job, _error) => {
      const {
        tokenId,
        endpoint,
        responseBody,
        responseStatus,
      } = (_error as any) as ExecutionError;

      await incidentForwarder?.dispatch(
        {
          endpoint,
          tokenId,
          payload: job.payload,
          id: job.id,
          runAt: job.runAt,
        },
        {
          body: responseBody,
          status: responseStatus,
        }
      );

      await telemetrist?.dispatch("execution_errored");
    }
  );

  async function close() {
    await worker.close();
    await redisClient.quit();
  }

  return {
    close,
  };
}

1	import { decodeQueueDescriptor } from "../shared/queue-descriptor";	10✔
2	import { UsageMeter } from "../shared/usage-meter";	10✔
3	import fetch from "cross-fetch";	10✔
4	import { TokenRepo } from "../shared/token-repo";	10✔
5	import { sign } from "secure-webhooks";	10✔
6	import { Redis } from "ioredis";
7	import { Telemetrist } from "../shared/telemetrist";	10✔
8	import { createOwl } from "../shared/owl";	10✔
9	import type { Logger } from "../shared/logger";
10	import { IncidentForwarder } from "../shared/incident-forwarder";	10✔
11
12	interface ExecutionError {
13	toString(): string;
14	endpoint: string;
15	tokenId: string;
16	responseBody: string;
17	responseStatus: number;
18	}
19
20	export function replaceLocalhostWithDockerHost(url: string): string {	10✔
21	if (url.startsWith("http://localhost")) {	1!
22	return url.replace("http://localhost", "http://host.docker.internal");	1✔
23	}
24
25	if (url.startsWith("https://localhost")) {	×
26	return url.replace("https://localhost", "https://host.docker.internal");	×
27	}
28
29	return url;	×
30	}
31
32	export interface QuirrelWorkerConfig {
33	redisFactory: () => Redis;
34	enableUsageMetering?: boolean;
35	runningInDocker?: boolean;
36	concurrency?: number;
37	disableTelemetry?: boolean;
38	logger?: Logger;
39	incidentReceiver?: { endpoint: string; passphrase: string };
40	}
41
42	export async function createWorker({	10✔
43	redisFactory,	23✔
44	enableUsageMetering,	23✔
45	runningInDocker,	23✔
46	concurrency = 100,	23!
47	disableTelemetry,	23✔
48	logger,	23✔
49	incidentReceiver,	23✔
50	}: QuirrelWorkerConfig) {
51	const incidentForwarder = incidentReceiver	23✔
52	? new IncidentForwarder(	23✔
53	incidentReceiver.endpoint,
54	incidentReceiver.passphrase
55	)
56	: undefined;
57
58	const redisClient = redisFactory();	23✔
59	const telemetrist = disableTelemetry	23✔
60	? undefined	23!
61	: new Telemetrist(runningInDocker ?? false);	×
62
63	const tokenRepo = new TokenRepo(redisClient);	23✔
64
65	let usageMeter: UsageMeter \| undefined = undefined;	23✔
66	if (enableUsageMetering) {	23✔
67	usageMeter = new UsageMeter(redisClient);	2✔
68	}
69
70	const owl = createOwl(redisFactory);	23✔
71
72	const worker = owl.createWorker(	23✔
73	async (job, jobMeta) => {	88✔
74	let { tokenId, endpoint } = decodeQueueDescriptor(job.queue);	88✔
75	const body = job.payload;	88✔
76
77	const executionDone = logger?.startingExecution({	88!
78	tokenId,
79	endpoint,
80	body,
81	id: job.id,
82	});
83
84	const headers: Record<string, string> = {	88✔
85	"Content-Type": "text/plain",
86	};
87
88	if (tokenId) {	88!
89	const token = await tokenRepo.getById(tokenId);	88✔
90	if (token) {	88✔
91	const signature = sign(body ?? "", token);	6!
92	headers["x-quirrel-signature"] = signature;	6✔
93	}
94	}
95
96	if (runningInDocker) {	88!
NEW 97	endpoint = replaceLocalhostWithDockerHost(endpoint);	×
98	}
99
100	const [response] = await Promise.all([	88✔
101	fetch(endpoint, {
102	method: "POST",
103	body,
104	headers,
105	}),
106	usageMeter?.record(tokenId),	264✔
107	]);
108
109	if (response.status >= 200 && response.status < 300) {	83✔
110	executionDone?.();	59!
111
112	telemetrist?.dispatch("dispatch_job");	59!
113	} else {
114	const responseBody = await response.text();	12✔
115
116	logger?.executionErrored(	12!
117	{ endpoint, tokenId, body: job.payload, id: job.id },
118	responseBody
119	);
120
121	if (response.status === 404) {	12✔
122	jobMeta.dontReschedule();	8✔
123	}
124
125	const error: ExecutionError = {	12✔
126	tokenId,
127	endpoint,
128	responseBody,
129	responseStatus: response.status,
130	toString() {
131	return responseBody;	12✔
132	},
133	};
134
135	throw error;	12✔
136	}
137	},
138	async (job, _error) => {	20✔
139	const {	20✔
140	tokenId,
141	endpoint,
142	responseBody,
143	responseStatus,
144	} = (_error as any) as ExecutionError;
145
146	await incidentForwarder?.dispatch(	20✔
147	{
148	endpoint,
149	tokenId,
150	payload: job.payload,
151	id: job.id,
152	runAt: job.runAt,
153	},
154	{
155	body: responseBody,
156	status: responseStatus,
157	}
158	);
159
160	await telemetrist?.dispatch("execution_errored");	20!
161	}
162	);
163
164	async function close() {
165	await worker.close();	40✔
166	await redisClient.quit();	34✔
167	}
168
169	return {	23✔
170	close,
171	};
172	}

quirrel-dev / quirrel / 536940436

Source File Press 'n' to go to next uncovered line, 'b' for previous

Source File
Press 'n' to go to next uncovered line, 'b' for previous