1import {
2 BatchRetryExhaustedError,
3 makeBatchedSink,
4} from "@/AWS/internal/BatchedSink.ts";
5import { describe, expect, it } from "alchemy-test";
6import * as Data from "effect/Data";
7import * as Effect from "effect/Effect";
8import * as Ref from "effect/Ref";
9import * as Schedule from "effect/Schedule";
10import * as Stream from "effect/Stream";
11
12/**
13 * Pure tests of the shared batch-sink engine (`AWS/internal/BatchedSink.ts`)
14 * with an injected `send`. Live end-to-end coverage of the three rebased
15 * sinks (SQS `QueueSink`, SNS `TopicSink`, Kinesis `StreamSink`) lives in
16 * their fixture tests; partial-failure classification cannot be forced
17 * deterministically against real AWS, so the retry semantics are proven here.
18 */
19
20class SendFailure extends Data.TaggedError("SendFailure")<{
21 readonly reason: string;
22}> {}
23
24interface FakeOut {
25 /** Indices (relative to the sent batch) reported as transiently unprocessed. */
26 readonly unprocessed?: readonly number[];
27 /** Indices (relative to the sent batch) reported as permanently rejected. */
28 readonly rejected?: readonly number[];
29}
30
31const byIndex = <In>(
32 indices: readonly number[] | undefined,
33 batch: readonly In[],
34): readonly In[] => {
35 if (indices === undefined || indices.length === 0) {
36 return [];
37 }
38 const set = new Set(indices);
39 return batch.filter((_, index) => set.has(index));
40};
41
42const instant = Schedule.recurs(3);
43
44describe("makeBatchedSink", { tags: ["unit", "provider:aws", "local"] }, () => {
45 it.effect("splits a chunk into maxRecords batches preserving order", () =>
46 Effect.gen(function* () {
47 const sent = yield* Ref.make<(readonly number[])[]>([]);
48 const sink = makeBatchedSink<number, FakeOut, never>({
49 maxRecords: 3,
50 send: (batch) =>
51 Ref.update(sent, (batches) => [...batches, batch]).pipe(
52 Effect.as({}),
53 ),
54 });
55
56 yield* Stream.make(1, 2, 3, 4, 5, 6, 7, 8).pipe(Stream.run(sink));
57
58 expect(yield* Ref.get(sent)).toEqual([
59 [1, 2, 3],
60 [4, 5, 6],
61 [7, 8],
62 ]);
63 }),
64 );
65
66 it.effect("packs greedily by maxBytes using sizeOf", () =>
67 Effect.gen(function* () {
68 const sent = yield* Ref.make<(readonly string[])[]>([]);
69 const sink = makeBatchedSink<string, FakeOut, never>({
70 maxRecords: 10,
71 maxBytes: 5,
72 sizeOf: (record) => record.length,
73 send: (batch) =>
74 Ref.update(sent, (batches) => [...batches, batch]).pipe(
75 Effect.as({}),
76 ),
77 });
78
79 // each record is 2 bytes; 3 records would be 6 > 5, so batches of 2
80 yield* Stream.make("aa", "bb", "cc", "dd", "ee").pipe(Stream.run(sink));
81
82 expect(yield* Ref.get(sent)).toEqual([
83 ["aa", "bb"],
84 ["cc", "dd"],
85 ["ee"],
86 ]);
87 }),
88 );
89
90 it.effect("re-submits only the unprocessed subset until it converges", () =>
91 Effect.gen(function* () {
92 const sent = yield* Ref.make<(readonly number[])[]>([]);
93 const sink = makeBatchedSink<number, FakeOut, never>({
94 maxRecords: 10,
95 retrySchedule: instant,
96 send: (batch) =>
97 Effect.gen(function* () {
98 const attempt = (yield* Ref.get(sent)).length;
99 yield* Ref.update(sent, (batches) => [...batches, batch]);
100 // first call: report the last two entries as unprocessed;
101 // second call: everything succeeds.
102 return attempt === 0
103 ? { unprocessed: [batch.length - 2, batch.length - 1] }
104 : {};
105 }),
106 unprocessed: (out, batch) => byIndex(out.unprocessed, batch),
107 });
108
109 yield* Stream.make(1, 2, 3, 4).pipe(Stream.run(sink));
110
111 expect(yield* Ref.get(sent)).toEqual([
112 [1, 2, 3, 4],
113 [3, 4],
114 ]);
115 }),
116 );
117
118 it.effect(
119 "fails with BatchRetryExhaustedError carrying the stranded entries",
120 () =>
121 Effect.gen(function* () {
122 const calls = yield* Ref.make(0);
123 const sink = makeBatchedSink<number, FakeOut, never>({
124 maxRecords: 10,
125 retrySchedule: Schedule.recurs(2),
126 send: (batch) =>
127 Ref.update(calls, (n) => n + 1).pipe(
128 // the last entry is reported unprocessed forever
129 Effect.as({ unprocessed: [batch.length - 1] }),
130 ),
131 unprocessed: (out, batch) => byIndex(out.unprocessed, batch),
132 });
133
134 const error = yield* Stream.make(1, 2, 3).pipe(
135 Stream.run(sink),
136 Effect.flip,
137 );
138
139 expect(error._tag).toBe("BatchRetryExhaustedError");
140 expect((error as BatchRetryExhaustedError<number>).entries).toEqual([
141 3,
142 ]);
143 // initial attempt + 2 bounded retries
144 expect(yield* Ref.get(calls)).toBe(3);
145 }),
146 { tags: ["provider:aws:batch"] },
147 );
148
149 it.effect("drops rejected entries without retrying and surfaces them", () =>
150 Effect.gen(function* () {
151 const calls = yield* Ref.make(0);
152 const rejected = yield* Ref.make<readonly number[]>([]);
153 const sink = makeBatchedSink<number, FakeOut, never>({
154 maxRecords: 10,
155 retrySchedule: instant,
156 send: () =>
157 Ref.update(calls, (n) => n + 1).pipe(
158 // the middle entry is permanently rejected; nothing is retried
159 Effect.as({ rejected: [1] } satisfies FakeOut),
160 ),
161 unprocessed: (out, batch) => byIndex(out.unprocessed, batch),
162 rejected: (out, batch) => byIndex(out.rejected, batch),
163 onRejected: (entries) => Ref.set(rejected, entries),
164 });
165
166 yield* Stream.make(1, 2, 3).pipe(Stream.run(sink));
167
168 expect(yield* Ref.get(calls)).toBe(1);
169 expect(yield* Ref.get(rejected)).toEqual([2]);
170 }),
171 );
172
173 it.effect("propagates send errors through the sink error channel", () =>
174 Effect.gen(function* () {
175 const sink = makeBatchedSink<number, FakeOut, SendFailure>({
176 maxRecords: 10,
177 send: () => Effect.fail(new SendFailure({ reason: "boom" })),
178 });
179
180 const error = yield* Stream.make(1).pipe(Stream.run(sink), Effect.flip);
181
182 expect(error._tag).toBe("SendFailure");
183 }),
184 );
185});