1/**
2 * `Hasher` over an AWS Lambda (DESIGN §22.11). Each chunk is one
3 * `InvokeFunction` call — a JSON payload with the chunk in base64 — signed
4 * from the Worker with the cross-cloud identity `AWS.Lambda.InvokeFunctionHttp`
5 * provisions for a Worker host (an IAM user, key and assume-role Role,
6 * once per Worker). Chunks are 4 MiB (the 6 MB invoke limit) and the pump
7 * writes the spill parts itself (`writesSpill: false`). Any failure on the
8 * Lambda side — throttling, a cold-start timeout, an oversize response —
9 * falls back to hashing that chunk inline, so a push never fails because
10 * the hasher did.
11 */
12import * as Effect from "effect/Effect";
13import * as Layer from "effect/Layer";
14import * as Stream from "effect/Stream";
15import { InvokeFunction } from "../../AWS/Lambda/InvokeFunction.ts";
16import type { Function as LambdaFunction } from "../../AWS/Lambda/Function.ts";
17import {
18 hashBounds,
19 resolveDeltas,
20 scanPart,
21} from "../Protocol/PartialScan.ts";
22import { Hasher, HashError, type HasherShape } from "./Hasher.ts";
23import {
24 decodeHashResponse,
25 encodeHashEvent,
26 LAMBDA_CHUNK_BYTES,
27 type HashResponse,
28} from "./LambdaEvent.ts";
29
30export const HasherLambda = (
31 fn: LambdaFunction | Effect.Effect<LambdaFunction, any, any>,
32): Layer.Layer<Hasher, never, InvokeFunction> =>
33 Layer.effect(
34 Hasher,
35 Effect.gen(function* () {
36 // A class-style declaration (`HasherFunction`) is yielded here — at
37 // deploy that registers the function, at runtime it resolves it.
38 const func = Effect.isEffect(fn)
39 ? yield* fn as Effect.Effect<LambdaFunction>
40 : fn;
41 const invoke = yield* InvokeFunction(func);
42 const remote = (
43 payload: Uint8Array,
44 options: Parameters<HasherShape["hashPart"]>[1],
45 ) =>
46 Effect.gen(function* () {
47 const response = yield* invoke({
48 Payload: JSON.stringify(encodeHashEvent(payload, options)),
49 }).pipe(
50 Effect.mapError(
51 (error) =>
52 new HashError({
53 reason: `lambda invoke: ${error._tag}${"message" in error ? `: ${String(error.message)}` : ""}`,
54 }),
55 ),
56 );
57 if (response.FunctionError !== undefined) {
58 return yield* new HashError({
59 reason: `lambda hasher: ${response.FunctionError}`,
60 });
61 }
62 const chunks = response.Payload
63 ? Array.from(
64 yield* Stream.runCollect(response.Payload).pipe(
65 Effect.mapError(
66 (error) =>
67 new HashError({
68 reason: `lambda response: ${String(error)}`,
69 }),
70 ),
71 ),
72 )
73 : [];
74 const text = new TextDecoder().decode(
75 chunks.length === 1 ? chunks[0] : concat(chunks),
76 );
77 const parsed = yield* Effect.try({
78 try: () => JSON.parse(text) as HashResponse,
79 catch: () => new HashError({ reason: `lambda response: not JSON` }),
80 });
81 return yield* decodeHashResponse(parsed);
82 });
83 return {
84 writesSpill: false,
85 chunkBytes: LAMBDA_CHUNK_BYTES,
86 hashPart: (payload, options) =>
87 // A region larger than a chunk (a big object straddling parts)
88 // cannot fit the invoke payload: hash it here without the trip.
89 (payload.length > LAMBDA_CHUNK_BYTES
90 ? Effect.fail(
91 new HashError({
92 reason: `payload of ${payload.length} bytes exceeds the invoke limit`,
93 }),
94 )
95 : remote(payload, options)
96 ).pipe(
97 Effect.catchTag("HashError", (error) => {
98 if (payload.length <= LAMBDA_CHUNK_BYTES) {
99 console.warn(`[hasher] ${error.reason}; hashing inline`);
100 }
101 const skip = options.skip ?? 0;
102 return scanPart(
103 skip > 0 ? payload.subarray(skip) : payload,
104 options,
105 );
106 }),
107 ),
108 // Delta batches stay on the receiver for now: a batch would need the
109 // same base64 event budget as a chunk.
110 resolveDeltas: (bases, jobs, options) =>
111 resolveDeltas(bases, jobs, options),
112 hashBoundsPart: (payload, bounds, options) =>
113 hashBounds(payload, bounds, options),
114 } satisfies HasherShape;
115 }),
116 );
117
118const concat = (chunks: ReadonlyArray<Uint8Array>): Uint8Array => {
119 const out = new Uint8Array(chunks.reduce((n, c) => n + c.length, 0));
120 let at = 0;
121 for (const c of chunks) {
122 out.set(c, at);
123 at += c.length;
124 }
125 return out;
126};