Merge pull request #276 from RhysSullivan/rs/secrets-parallel-fanout

perf(secrets): parallelize provider fan-out and header resolution

Rhys Sullivan committed 6 months ago

Commit 36ef8382 parents 370df07 165e7703 files · +100 −51Browse files

Unified · Split

3 files

packages/core/sdk/src/executor.ts +45 −25
packages/plugins/graphql/src/sdk/invoke.ts +28 −11
packages/plugins/openapi/src/sdk/invoke.ts +27 −15

           return yield* provider.get(id);
         }
-        // Fallback: ask enumerating providers in registration order.
+        // Fallback: ask every enumerating provider in parallel. First
-        // First non-null wins. Providers that throw are treated as
+        // non-null in registration order wins. Providers that throw
-        // "don't have it" and skipped so one flaky provider doesn't
+        // are treated as "don't have it" so one flaky provider can't
         // block resolution via others.
-        for (const provider of secretProviders.values()) {
+        const candidates = [...secretProviders.values()].filter(
-          if (!provider.list) continue;
+          (p) => p.list,
-          const value = yield* provider
+        );
-            .get(id)
+        const values = yield* Effect.all(
-            .pipe(Effect.catchAll(() => Effect.succeed(null)));
+          candidates.map((p) =>
-          if (value !== null) return value;
+            p.get(id).pipe(Effect.catchAll(() => Effect.succeed(null))),
-        }
+          ),
+          { concurrency: "unbounded" },
+        );
+        for (const value of values) if (value !== null) return value;
         return null;
       });
     const secretsRemove = (id: string): Effect.Effect<void, Error> =>
       Effect.gen(function* () {
-        for (const provider of secretProviders.values()) {
+        // Providers don't coordinate on which of them own the id — they
-          if (provider.writable && provider.delete) {
+        // each get asked. Most calls are no-ops; fan them out so one
-            yield* provider.delete(id);
+        // slow provider doesn't serialize the rest.
-          }
+        const deleters = [...secretProviders.values()].filter(
-        }
+          (p): p is typeof p & { delete: NonNullable<typeof p.delete> } =>
+            !!(p.writable && p.delete),
+        );
+        yield* Effect.all(
+          deleters.map((p) => p.delete(id)),
+          { concurrency: "unbounded" },
+        );
         yield* core.delete({
           model: "secret",
           where: [{ field: "id", value: id }],
           );
         }
-        // Then every provider that can enumerate itself. If a provider
+        // Then every provider that can enumerate itself, in parallel.
-        // fails to list (unlocked vault, network error), swallow the
+        // If a provider fails to list (unlocked vault, network error),
-        // failure and continue — one flaky provider shouldn't block
+        // swallow the failure so one flaky provider can't block the
-        // the whole list.
+        // whole list. Merge in registration order afterwards so the
-        for (const [providerKey, provider] of secretProviders.entries()) {
+        // "first provider wins" precedence stays deterministic.
-          if (!provider.list) continue;
+        const listers = [...secretProviders.entries()].filter(
-          const entries = yield* provider
+          ([, p]) => p.list,
-            .list()
+        );
-            .pipe(Effect.catchAll(() => Effect.succeed([] as const)));
+        const lists = yield* Effect.all(
+          listers.map(([key, p]) =>
+            p
+              .list!()
+              .pipe(
+                Effect.catchAll(() => Effect.succeed([] as const)),
+                Effect.map((entries) => ({ key, entries })),
+              ),
+          ),
+          { concurrency: "unbounded" },
+        );
+        for (const { key, entries } of lists) {
           for (const entry of entries) {
             if (byId.has(entry.id)) continue; // core row wins
             byId.set(
                 id: SecretId.make(entry.id),
                 scopeId: scope.id,
                 name: entry.name,
-                provider: providerKey,
+                provider: key,
                 createdAt: new Date(),
               }),
             );

packages/plugins/graphql/src/sdk/invoke.ts+28−11

@@ -17,18 +17,35 @@ export const resolveHeaders = ( · 16 unmodified lines
17	secrets: { readonly get: (id: string) => Effect.Effect<string \| null, Error> },	17	secrets: { readonly get: (id: string) => Effect.Effect<string \| null, Error> },
18	): Effect.Effect<Record<string, string>, Error> =>	18	): Effect.Effect<Record<string, string>, Error> =>
19	Effect.gen(function* () {	19	Effect.gen(function* () {
		20	const entries = Object.entries(headers);
		21	// Resolve secret-backed headers in parallel. Missing / failing
		22	// lookups drop the header rather than fail the invocation, same
		23	// as the serial version.
		24	const values = yield* Effect.all(
		25	entries.map(([name, value]) =>
		26	typeof value === "string"
		27	? Effect.succeed<{ readonly name: string; readonly value: string \| null }>({
		28	name,
		29	value,
		30	})
		31	: secrets.get(value.secretId).pipe(
		32	Effect.catchAll(() => Effect.succeed<string \| null>(null)),
		33	Effect.map((secret) => ({
		34	name,
		35	value:
		36	secret === null
		37	? null
		38	: value.prefix
		39	? `${value.prefix}${secret}`
		40	: secret,
		41	})),
		42	),
		43	),
		44	{ concurrency: "unbounded" },
		45	);
20	const resolved: Record<string, string> = {};	46	const resolved: Record<string, string> = {};
21	for (const [name, value] of Object.entries(headers)) {	47	for (const { name, value } of values) {
22	if (typeof value === "string") {	48	if (value !== null) resolved[name] = value;
23	resolved[name] = value;
24	} else {
25	const secret = yield* secrets.get(value.secretId).pipe(
26	Effect.catchAll(() => Effect.succeed<string \| null>(null)),
27	);
28	if (secret !== null) {
29	resolved[name] = value.prefix ? `${value.prefix}${secret}` : secret;
30	}
31	}
32	}	49	}
33	return resolved;	50	return resolved;
34	});	51	});

packages/plugins/openapi/src/sdk/invoke.ts+27−15

@@ -95,22 +95,34 @@ export const resolveHeaders = ( · 94 unmodified lines
95	secrets: { readonly get: (id: string) => Effect.Effect<string \| null, Error> },	95	secrets: { readonly get: (id: string) => Effect.Effect<string \| null, Error> },
96	): Effect.Effect<Record<string, string>, Error> =>	96	): Effect.Effect<Record<string, string>, Error> =>
97	Effect.gen(function* () {	97	Effect.gen(function* () {
98	const resolved: Record<string, string> = {};	98	const entries = Object.entries(headers);
99	for (const [name, value] of Object.entries(headers)) {	99	// Fan out secret lookups: on every invocation, one or two headers
100	if (typeof value === "string") {	100	// typically each hit the secret store. Resolving them in parallel
101	resolved[name] = value;	101	// is a free wall-clock win — preserved order is only needed for
102	} else {	102	// the final assembly, not the fetches.
103	const secret = yield* secrets.get(value.secretId);	103	const values = yield* Effect.all(
104	if (secret === null) {	104	entries.map(([name, value]) =>
105	return yield* Effect.fail(	105	typeof value === "string"
106	new Error(	106	? Effect.succeed({ name, value })
107	`Failed to resolve secret "${value.secretId}" for header "${name}"`,	107	: secrets.get(value.secretId).pipe(
		108	Effect.flatMap((secret) =>
		109	secret === null
		110	? Effect.fail(
		111	new Error(
		112	`Failed to resolve secret "${value.secretId}" for header "${name}"`,
		113	),
		114	)
		115	: Effect.succeed({
		116	name,
		117	value: value.prefix ? `${value.prefix}${secret}` : secret,
		118	}),
		119	),
108	),	120	),
109	);	121	),
110	}	122	{ concurrency: "unbounded" },
111	resolved[name] = value.prefix ? `${value.prefix}${secret}` : secret;	123	);
112	}	124	const resolved: Record<string, string> = {};
113	}	125	for (const { name, value } of values) resolved[name] = value;
114	return resolved;	126	return resolved;
115	});	127	});
116		128