1import * as Cloudflare from "@/Cloudflare";
2import { CloudflareEnvironment } from "@/Cloudflare/CloudflareEnvironment";
3import type { SubscriptionResourceSource } from "@/Cloudflare/Queues/Subscription";
4import * as TestCore from "@/Test/Core";
5import * as Test from "@/Test/Alchemy";
6import * as ai from "@distilled.cloud/cloudflare/ai";
7import * as kv from "@distilled.cloud/cloudflare/kv";
8import * as images from "@distilled.cloud/cloudflare/images";
9import * as queues from "@distilled.cloud/cloudflare/queues";
10import * as r2 from "@distilled.cloud/cloudflare/r2";
11import * as vectorize from "@distilled.cloud/cloudflare/vectorize";
12import * as workers from "@distilled.cloud/cloudflare/workers";
13import { describe, expect, test as unitTest } from "alchemy-test";
14import * as Effect from "effect/Effect";
15import * as Exit from "effect/Exit";
16import * as Fiber from "effect/Fiber";
17import * as TestClock from "effect/testing/TestClock";
18import * as Layer from "effect/Layer";
19import * as Cause from "effect/Cause";
20import * as Clock from "effect/Clock";
21import * as Schedule from "effect/Schedule";
22import * as Schema from "effect/Schema";
23import {
24 hasReadySubscriptionEvent,
25 makeSubscriptionCleanup,
26 matchesSubscriptionEvent,
27 SubscriptionEvent,
28 type SubscriptionProbe,
29} from "./SubscriptionSources.ts";
30
31const { test } = Test.make({ providers: Cloudflare.providers() });
32
33type Assert<T extends true> = T;
34type Accepts<T> = T extends SubscriptionResourceSource ? true : false;
35type _Variant = Assert<Accepts<Cloudflare.Images.Variant>>;
36type _Namespace = Assert<Accepts<Cloudflare.KV.Namespace>>;
37type _Bucket = Assert<Accepts<Cloudflare.R2.Bucket>>;
38type _Job = Assert<Accepts<Cloudflare.R2.SuperSlurperJob>>;
39type _Index = Assert<Accepts<Cloudflare.Vectorize.Index>>;
40type _Model = Assert<Accepts<Cloudflare.AI.Model>>;
41type _SearchModel = Assert<
42 Cloudflare.AI.SearchModel extends string ? true : false
43>;
44type _SearchModelIsNotResource = Assert<
45 Accepts<Cloudflare.AI.SearchModel> extends false ? true : false
46>;
47type _Worker = Assert<Accepts<Cloudflare.Worker>>;
48
49const references = Effect.gen(function* () {
50 const queue = yield* Cloudflare.Queues.Queue("EventsQueue");
51 const sources = [
52 yield* Cloudflare.Images.Variant.ref("Image"),
53 yield* Cloudflare.KV.Namespace.ref("Cache"),
54 yield* Cloudflare.R2.Bucket.ref("Uploads"),
55 yield* Cloudflare.R2.SuperSlurperJob.ref("Migration"),
56 yield* Cloudflare.Vectorize.Index.ref("Search"),
57 yield* Cloudflare.AI.Model.ref("Embeddings"),
58 yield* Cloudflare.Worker.ref("Website"),
59 ];
60 const Subscription = yield* Cloudflare.Queues.Subscription;
61 for (const [index, source] of sources.entries()) {
62 yield* Subscription(
63 `Events${index}`,
64 Effect.succeed({
65 source,
66 events: ["product.event"],
67 queueId: queue.queueId,
68 }),
69 );
70 }
71 yield* Subscription("Invalid", {
72 // @ts-expect-error arbitrary account-bearing objects are not resource sources
73 source: { accountId: "account", modelName: "model" },
74 events: ["batch.queued"],
75 queueId: queue.queueId,
76 });
77});
78void references;
79
80type Kind = "images" | "kv" | "r2" | "vectorize" | "model" | "worker";
81const createSource = (kind: Kind, id = "Source") =>
82 Effect.gen(function* () {
83 switch (kind) {
84 case "images":
85 return yield* Cloudflare.Images.Variant(id, {
86 name: `alchemySubscription${id}`,
87 fit: "contain",
88 width: 100,
89 height: 100,
90 });
91 case "kv":
92 return yield* Cloudflare.KV.Namespace(id);
93 case "r2":
94 return yield* Cloudflare.R2.Bucket(id, { forceDestroy: true });
95 case "vectorize":
96 return yield* Cloudflare.Vectorize.Index(id, {
97 dimensions: 32,
98 metric: "cosine",
99 });
100 case "model":
101 return yield* Cloudflare.AI.Model(id, {
102 modelName: "@cf/baai/bge-m3",
103 });
104 case "worker":
105 return yield* Cloudflare.Worker(id, {
106 main: `${import.meta.dirname}/fixtures/subscription-source-worker.ts`,
107 });
108 }
109 });
110const refSource = (kind: Kind, options?: { stack: string; stage: string }) =>
111 Effect.gen(function* () {
112 switch (kind) {
113 case "images":
114 return yield* Cloudflare.Images.Variant.ref("Source", options);
115 case "kv":
116 return yield* Cloudflare.KV.Namespace.ref("Source", options);
117 case "r2":
118 return yield* Cloudflare.R2.Bucket.ref("Source", options);
119 case "vectorize":
120 return yield* Cloudflare.Vectorize.Index.ref("Source", options);
121 case "model":
122 return yield* Cloudflare.AI.Model.ref("Source", options);
123 case "worker":
124 return yield* Cloudflare.Worker.ref("Source", options);
125 }
126 });
127const eventType = (kind: Kind) =>
128 ({
129 images: process.env.CLOUDFLARE_TEST_IMAGES_EVENT ?? "image.uploaded",
130 kv: "namespace.created",
131 r2: "bucket.created",
132 vectorize: "index.created",
133 model: "batch.queued",
134 worker: "build.started",
135 })[kind];
136
137const verifySource = (source: {
138 accountId: string;
139 variantName?: string;
140 namespaceId?: string;
141 bucketName?: string;
142 indexName?: string;
143 modelName?: string;
144 workerName?: string;
145}) =>
146 Effect.gen(function* () {
147 const { accountId } = source;
148 if (source.variantName)
149 yield* images.getV1Variant({ accountId, variantId: source.variantName });
150 else if (source.namespaceId)
151 yield* kv.getNamespace({ accountId, namespaceId: source.namespaceId });
152 else if (source.bucketName)
153 yield* r2.getBucket({ accountId, bucketName: source.bucketName });
154 else if (source.indexName)
155 yield* vectorize.getIndex({ accountId, indexName: source.indexName });
156 else if (source.modelName)
157 yield* ai.getModelSchema({ accountId, model: source.modelName });
158 else if (source.workerName)
159 yield* workers.getScriptScriptAndVersionSetting({
160 accountId,
161 scriptName: source.workerName,
162 });
163 else throw new Error("Unknown source identity");
164 });
165
166const gone = (accountId: string, subscriptionId: string) =>
167 queues.getSubscription({ accountId, subscriptionId }).pipe(
168 Effect.flatMap(() => Effect.fail(new Error("Subscription still exists"))),
169 Effect.catchTag("SubscriptionNotFound", () => Effect.void),
170 );
171
172const triggerEvent = (kind: Kind, accountId: string, name: string) =>
173 Effect.gen(function* () {
174 const eventName = `${name.slice(0, 48)}-event`;
175 if (kind === "kv") {
176 const namespace = yield* Effect.acquireRelease(
177 kv.createNamespace({ accountId, title: eventName }),
178 (namespace) =>
179 kv
180 .deleteNamespace({ accountId, namespaceId: namespace.id })
181 .pipe(Effect.orDie),
182 );
183 return namespace.id;
184 }
185 if (kind === "r2") {
186 const bucket = yield* Effect.acquireRelease(
187 r2.createBucket({ accountId, name: eventName }),
188 () =>
189 r2
190 .deleteBucket({ accountId, bucketName: eventName })
191 .pipe(Effect.orDie),
192 );
193 return bucket.name!;
194 }
195 if (kind === "vectorize") {
196 const index = yield* Effect.acquireRelease(
197 vectorize.createIndex({
198 accountId,
199 name: eventName,
200 config: { dimensions: 32, metric: "cosine" },
201 }),
202 () =>
203 vectorize
204 .deleteIndex({ accountId, indexName: eventName })
205 .pipe(Effect.orDie),
206 );
207 yield* Effect.logInfo("Vectorize index created", {
208 name: index.name,
209 createdOn: index.createdOn,
210 });
211 return index.name!;
212 }
213 return yield* Effect.fail(new Error(`No lifecycle trigger for ${kind}`));
214 });
215
216const pullEvents = (
217 accountId: string,
218 queueId: string,
219 received: SubscriptionEvent[] = [],
220) =>
221 Effect.gen(function* () {
222 const batch = yield* queues
223 .pullMessage({
224 accountId,
225 queueId,
226 batchSize: 100,
227 visibilityTimeoutMs: 30_000,
228 })
229 .pipe(
230 Effect.retry({
231 while: (error) => error._tag === "QueueHttpPullNotEnabled",
232 schedule: Schedule.spaced("2 seconds"),
233 times: 8,
234 }),
235 );
236 const events = yield* Effect.forEach(
237 (batch.messages ?? []).flatMap(({ body }) => (body ? [body] : [])),
238 (body) => Schema.decodeUnknownEffect(SubscriptionEvent)(body),
239 );
240 received.push(...events);
241 if (events.length) {
242 yield* Effect.logInfo("Subscription events received", events);
243 }
244 const acks = (batch.messages ?? []).flatMap(({ leaseId }) =>
245 leaseId ? [{ leaseId }] : [],
246 );
247 if (acks.length) {
248 const result = yield* queues.ackMessage({ accountId, queueId, acks });
249 expect(result.ackCount).toBe(acks.length);
250 expect(Object.keys(result.warnings ?? {})).toHaveLength(0);
251 }
252 return events;
253 });
254
255const waitForDelivery = (
256 kind: "kv" | "r2" | "vectorize",
257 accountId: string,
258 queueId: string,
259 subscriptionId: string,
260 readyAfter: number,
261) =>
262 Effect.gen(function* () {
263 const probes: SubscriptionProbe[] = [];
264 let ready = false;
265 const createProbe = Effect.gen(function* () {
266 const createdAt = yield* Clock.currentTimeMillis;
267 probes.push({
268 createdAt,
269 identity: yield* triggerEvent(
270 kind,
271 accountId,
272 `${kind}-${subscriptionId}-${probes.length}`,
273 ),
274 });
275 });
276 const observe = Effect.gen(function* () {
277 const events = yield* pullEvents(accountId, queueId);
278 ready = hasReadySubscriptionEvent(events, probes, {
279 source: kind,
280 type: eventType(kind),
281 accountId,
282 subscriptionId,
283 readyAfter,
284 });
285 });
286 if (kind === "vectorize") {
287 const now = yield* Clock.currentTimeMillis;
288 yield* Effect.sleep(Math.max(0, readyAfter - now));
289 yield* createProbe;
290 yield* observe.pipe(
291 Effect.repeat({
292 schedule: Schedule.spaced("2 seconds"),
293 times: 8,
294 until: () => ready,
295 }),
296 );
297 } else {
298 yield* createProbe.pipe(
299 Effect.andThen(observe),
300 Effect.repeat({
301 schedule: Schedule.spaced("7 seconds"),
302 times: 8,
303 until: () => ready,
304 }),
305 );
306 }
307 expect(ready).toBe(true);
308 }).pipe(Effect.timeout("80 seconds"));
309
310// Account-wide subscriptions are unique per product, regardless of the selected resource.
311describe.sequential(
312 "resource subscription sources",
313 {
314 tags: [
315 "provider:cloudflare",
316 "provider:cloudflare:kv",
317 "provider:cloudflare:queue",
318 "live",
319 ],
320 },
321 () => {
322 for (const kind of [
323 "images",
324 "kv",
325 "r2",
326 "vectorize",
327 "model",
328 "worker",
329 ] as const) {
330 test.provider(
331 `${kind} direct source and persisted ref retain identity and ownership`,
332 (stack) =>
333 Effect.gen(function* () {
334 yield* stack.destroy();
335 const { accountId } = yield* yield* CloudflareEnvironment;
336 const program = (reference: boolean) =>
337 Effect.gen(function* () {
338 const resource = yield* createSource(kind);
339 const queue = yield* Cloudflare.Queues.Queue("EventsQueue");
340 const subscription = yield* Cloudflare.Queues.Subscription(
341 "Events",
342 Effect.gen(function* () {
343 return {
344 source: reference ? yield* refSource(kind) : resource,
345 events: [eventType(kind)],
346 queueId: queue.queueId,
347 };
348 }),
349 );
350 return { resource, queue, subscription };
351 });
352 const initial = yield* stack.deploy(program(false));
353 const configuredAt = yield* Clock.currentTimeMillis;
354 expect(initial.subscription.accountId).toBe(accountId);
355 const expected =
356 kind === "model"
357 ? { type: "workersAi.model", modelName: "@cf/baai/bge-m3" }
358 : kind === "worker" && "workerName" in initial.resource
359 ? {
360 type: "workersBuilds.worker",
361 workerName: initial.resource.workerName,
362 }
363 : { type: kind };
364 expect(initial.subscription.source).toEqual(expected);
365 const plan = yield* stack.plan(program(false));
366 expect(plan.resources.Source.downstream).toContain("Events");
367 expect(plan.resources.Events.state).toHaveProperty(
368 "props.sourceAccountId",
369 accountId,
370 );
371 expect(plan.resources.Events.state).toHaveProperty(
372 "props.source",
373 expected,
374 );
375 const observed = yield* queues.getSubscription({
376 accountId,
377 subscriptionId: initial.subscription.subscriptionId,
378 });
379 expect(observed.source).toEqual(expect.objectContaining(expected));
380 const receivesLifecycle =
381 kind === "kv" || kind === "r2" || kind === "vectorize";
382 if (receivesLifecycle) {
383 yield* queues.createConsumer({
384 accountId,
385 queueId: initial.queue.queueId,
386 type: "http_pull",
387 });
388 yield* pullEvents(accountId, initial.queue.queueId);
389 }
390 const referenced = yield* stack.deploy(program(true));
391 expect(referenced.subscription.subscriptionId).toBe(
392 initial.subscription.subscriptionId,
393 );
394 if (receivesLifecycle) {
395 yield* Effect.gen(function* () {
396 yield* waitForDelivery(
397 kind,
398 accountId,
399 referenced.queue.queueId,
400 referenced.subscription.subscriptionId,
401 // Allow the observed Vectorize propagation interval before probing.
402 kind === "vectorize" ? configuredAt + 60_000 : 0,
403 );
404 const identities = [
405 yield* triggerEvent(
406 kind,
407 accountId,
408 referenced.queue.queueName,
409 ),
410 ];
411 if (kind === "vectorize") {
412 for (const suffix of ["second", "third"]) {
413 identities.push(
414 yield* triggerEvent(
415 kind,
416 accountId,
417 `vec-${referenced.subscription.subscriptionId}-${suffix}`,
418 ),
419 );
420 }
421 }
422 const events: SubscriptionEvent[] = [];
423 const missing = () =>
424 identities.filter(
425 (identity) =>
426 !events.some((event) =>
427 matchesSubscriptionEvent(event, {
428 source: kind,
429 type: eventType(kind),
430 accountId,
431 subscriptionId:
432 referenced.subscription.subscriptionId,
433 identity,
434 }),
435 ),
436 );
437 yield* pullEvents(
438 accountId,
439 referenced.queue.queueId,
440 events,
441 ).pipe(
442 Effect.repeat({
443 schedule: Schedule.spaced("5 seconds"),
444 times: 10,
445 until: () => missing().length === 0,
446 }),
447 Effect.timeout("60 seconds"),
448 Effect.onExit(() =>
449 Effect.gen(function* () {
450 if (missing().length === 0) return;
451 yield* Effect.logInfo("Missing subscription delivery", {
452 expected: identities,
453 missing: missing(),
454 received: events,
455 });
456 yield* Effect.all({
457 subscription: queues.getSubscription({
458 accountId,
459 subscriptionId:
460 referenced.subscription.subscriptionId,
461 }),
462 queue: queues.getQueue({
463 accountId,
464 queueId: referenced.queue.queueId,
465 }),
466 metrics: queues.getMetricsQueue({
467 accountId,
468 queueId: referenced.queue.queueId,
469 }),
470 }).pipe(
471 Effect.tap((state) =>
472 Effect.logInfo("Subscription delivery state", state),
473 ),
474 Effect.timeout("5 seconds"),
475 Effect.catch((error) =>
476 Effect.logWarning(
477 "Subscription delivery state unavailable",
478 { error: error._tag },
479 ),
480 ),
481 );
482 }),
483 ),
484 );
485 expect(missing()).toEqual([]);
486 }).pipe(Effect.timeout("90 seconds"));
487 }
488 yield* stack.deploy(createSource(kind));
489 yield* gone(accountId, initial.subscription.subscriptionId);
490 yield* verifySource(initial.resource);
491 yield* stack.destroy();
492 if (kind === "model")
493 yield* ai.getModelSchema({ accountId, model: "@cf/baai/bge-m3" });
494 }).pipe(
495 Effect.scoped,
496 Effect.ensuring(stack.destroy().pipe(Effect.orDie)),
497 ),
498 {
499 tags: [
500 "provider:cloudflare:ai",
501 "provider:cloudflare:images",
502 "provider:cloudflare:r2",
503 "provider:cloudflare:vectorize",
504 "provider:cloudflare:worker",
505 ],
506 timeout: 120_000,
507 exclusive: true,
508 },
509 );
510 }
511
512 test.provider(
513 "cross-stack and cross-stage namespace ref does not own the source",
514 (stack) => {
515 const host = TestCore.scratchStack(
516 { providers: Cloudflare.providers(), stage: `${stack.stage}-host` },
517 "SubscriptionRefHost",
518 "test/Cloudflare/Queue/SubscriptionSources.test.ts",
519 );
520 return Effect.gen(function* () {
521 yield* stack.destroy();
522 yield* host.destroy();
523 const source = yield* host.deploy(Cloudflare.KV.Namespace("Source"));
524 const program = Effect.gen(function* () {
525 const queue = yield* Cloudflare.Queues.Queue("Queue");
526 return yield* Cloudflare.Queues.Subscription("Events", {
527 source: yield* refSource("kv", {
528 stack: host.name,
529 stage: host.stage,
530 }),
531 events: ["namespace.created"],
532 queueId: queue.queueId,
533 });
534 });
535 const subscription = yield* stack.deploy(program);
536 expect(subscription.source).toEqual({ type: "kv" });
537 const plan = yield* stack.plan(program);
538 expect(
539 Object.values(plan.resources).some(
540 (node) => node.resource.Type === "Cloudflare.KV.Namespace",
541 ),
542 ).toBe(false);
543 yield* stack.destroy();
544 yield* verifySource(source);
545 yield* host.destroy();
546 }).pipe(
547 Effect.ensuring(
548 stack.destroy().pipe(Effect.andThen(host.destroy()), Effect.orDie),
549 ),
550 );
551 },
552 {
553 tags: [
554 "provider:cloudflare:ai",
555 "provider:cloudflare:images",
556 "provider:cloudflare:r2",
557 "provider:cloudflare:vectorize",
558 "provider:cloudflare:worker",
559 ],
560 timeout: 120_000,
561 exclusive: true,
562 },
563 );
564
565 test.provider(
566 "a source reference from another account is rejected before subscription creation",
567 (stack) => {
568 const host = TestCore.scratchStack(
569 { providers: Cloudflare.providers(), stage: `${stack.stage}-host` },
570 "SubscriptionAccountHost",
571 "test/Cloudflare/Queue/SubscriptionSources.test.ts",
572 );
573 return Effect.gen(function* () {
574 yield* stack.destroy();
575 yield* host.destroy();
576 const environment = yield* yield* CloudflareEnvironment;
577 const hosted = yield* host.deploy(
578 Effect.gen(function* () {
579 const source = yield* Cloudflare.KV.Namespace("Source");
580 const queue = yield* Cloudflare.Queues.Queue("Queue");
581 return { source, queue };
582 }),
583 );
584 const rejecting = TestCore.scratchStack(
585 {
586 providers: Layer.mergeAll(
587 Cloudflare.providers(),
588 Layer.succeed(
589 CloudflareEnvironment,
590 Effect.succeed({
591 ...environment,
592 accountId: "00000000000000000000000000000000",
593 }),
594 ),
595 ),
596 stage: stack.stage,
597 },
598 "SubscriptionAccountMismatch",
599 "test/Cloudflare/Queue/SubscriptionSources.test.ts",
600 );
601 const cleanup = TestCore.scratchStack(
602 { providers: Cloudflare.providers(), stage: stack.stage },
603 "SubscriptionAccountMismatch",
604 "test/Cloudflare/Queue/SubscriptionSources.test.ts",
605 );
606 yield* cleanup.destroy();
607 yield* Effect.addFinalizer(() =>
608 cleanup.destroy().pipe(Effect.orDie),
609 );
610 const mismatch = yield* rejecting
611 .deploy(
612 Effect.gen(function* () {
613 return yield* Cloudflare.Queues.Subscription("Events", {
614 source: yield* refSource("kv", {
615 stack: host.name,
616 stage: host.stage,
617 }),
618 events: ["namespace.created"],
619 queueId: hosted.queue.queueId,
620 });
621 }),
622 )
623 .pipe(Effect.exit);
624 expect(Exit.isFailure(mismatch)).toBe(true);
625 if (Exit.isFailure(mismatch))
626 expect(Cause.pretty(mismatch.cause)).toContain(
627 "SubscriptionSourceAccountMismatch",
628 );
629 yield* stack.destroy();
630 yield* verifySource(hosted.source);
631 yield* host.destroy();
632 }).pipe(
633 Effect.scoped,
634 Effect.ensuring(
635 stack.destroy().pipe(Effect.andThen(host.destroy()), Effect.orDie),
636 ),
637 );
638 },
639 {
640 tags: [
641 "provider:cloudflare:ai",
642 "provider:cloudflare:images",
643 "provider:cloudflare:r2",
644 "provider:cloudflare:vectorize",
645 "provider:cloudflare:worker",
646 ],
647 timeout: 120_000,
648 exclusive: true,
649 },
650 );
651
652 test.provider(
653 "a foreign subscription collision preserves the original destination",
654 (stack) => {
655 const host = TestCore.scratchStack(
656 { providers: Cloudflare.providers(), stage: `${stack.stage}-owner` },
657 "SubscriptionOwner",
658 "test/Cloudflare/Queue/SubscriptionSources.test.ts",
659 );
660 return Effect.gen(function* () {
661 yield* stack.destroy();
662 yield* host.destroy();
663 const owner = yield* host.deploy(
664 Effect.gen(function* () {
665 const queue = yield* Cloudflare.Queues.Queue("Queue");
666 const subscription = yield* Cloudflare.Queues.Subscription(
667 "OwnedEvents",
668 {
669 source: { type: "kv" },
670 events: ["namespace.created"],
671 queueId: queue.queueId,
672 },
673 );
674 return { queue, subscription };
675 }),
676 );
677 const result = yield* stack
678 .deploy(
679 Effect.gen(function* () {
680 const queue = yield* Cloudflare.Queues.Queue("Queue");
681 const source = yield* Cloudflare.KV.Namespace("Source");
682 return yield* Cloudflare.Queues.Subscription("ForeignEvents", {
683 source,
684 events: ["namespace.created"],
685 queueId: queue.queueId,
686 });
687 }),
688 )
689 .pipe(Effect.exit);
690 expect(Exit.isFailure(result)).toBe(true);
691 if (Exit.isFailure(result))
692 expect(Cause.pretty(result.cause)).toContain(
693 "SubscriptionAlreadyExists",
694 );
695 yield* stack.destroy();
696 const observed = yield* queues.getSubscription({
697 accountId: owner.subscription.accountId,
698 subscriptionId: owner.subscription.subscriptionId,
699 });
700 expect(observed.destination.queueId).toBe(owner.queue.queueId);
701 expect(observed.name).toBe(owner.subscription.name);
702 yield* host.destroy();
703 }).pipe(
704 Effect.ensuring(
705 stack.destroy().pipe(Effect.andThen(host.destroy()), Effect.orDie),
706 ),
707 );
708 },
709 { timeout: 120_000, exclusive: true },
710 );
711 },
712);
713
714describe(
715 "subscription cleanup",
716 {
717 tags: ["unit", "provider:cloudflare", "provider:cloudflare:queue", "local"],
718 },
719 () => {
720 unitTest.effect("returns successful cleanup results", () =>
721 Effect.gen(function* () {
722 const cleanup = makeSubscriptionCleanup();
723 expect(yield* cleanup(Effect.succeed(42))).toBe(42);
724 }),
725 );
726
727 unitTest.effect("preserves cleanup failures", () =>
728 Effect.gen(function* () {
729 const cleanup = makeSubscriptionCleanup();
730 const result = yield* cleanup(
731 Effect.fail(new Error("delete failed")),
732 ).pipe(Effect.exit);
733 expect(Exit.isFailure(result)).toBe(true);
734 if (Exit.isFailure(result))
735 expect(Cause.pretty(result.cause)).toContain("delete failed");
736 }),
737 );
738
739 unitTest.effect("bounds a stalled release inside scope finalization", () =>
740 Effect.gen(function* () {
741 const cleanup = makeSubscriptionCleanup();
742 const fiber = yield* Effect.acquireRelease(Effect.void, () =>
743 cleanup(Effect.never),
744 ).pipe(
745 Effect.scoped,
746 Effect.exit,
747 Effect.forkChild({ startImmediately: true }),
748 );
749 yield* TestClock.adjust("5 seconds");
750 const result = yield* Fiber.join(fiber);
751 expect(Exit.isFailure(result)).toBe(true);
752 if (Exit.isFailure(result))
753 expect(Cause.pretty(result.cause)).toContain("TimeoutError");
754 }),
755 );
756
757 unitTest.effect(
758 "shares one deadline across releases and stops starting work when exhausted",
759 () =>
760 Effect.gen(function* () {
761 const cleanup = makeSubscriptionCleanup();
762 yield* cleanup(Effect.void);
763 yield* TestClock.adjust("19 seconds");
764 const fiber = yield* cleanup(Effect.never).pipe(
765 Effect.exit,
766 Effect.forkChild({ startImmediately: true }),
767 );
768 yield* TestClock.adjust("1 second");
769 expect(Exit.isFailure(yield* Fiber.join(fiber))).toBe(true);
770 let started = false;
771 const result = yield* cleanup(
772 Effect.sync(() => {
773 started = true;
774 }),
775 ).pipe(Effect.exit);
776 expect(Exit.isFailure(result)).toBe(true);
777 expect(started).toBe(false);
778 if (Exit.isFailure(result))
779 expect(Cause.pretty(result.cause)).toContain(
780 "Subscription cleanup deadline exceeded",
781 );
782 }),
783 );
784 },
785);
786
787const expected = {
788 source: "vectorize" as const,
789 type: "index.created",
790 accountId: "account",
791 subscriptionId: "subscription",
792 identity: "vectorize-subscription-0-event",
793};
794const event: SubscriptionEvent = {
795 type: "cf.vectorize.index.created",
796 source: { type: "vectorize" },
797 metadata: {
798 accountId: "account",
799 eventSubscriptionId: "subscription",
800 eventTimestamp: "2026-09-20T00:00:00Z",
801 },
802 payload: { name: expected.identity },
803};
804
805describe(
806 "subscription event identity",
807 {
808 tags: ["unit", "provider:cloudflare", "provider:cloudflare:queue", "local"],
809 },
810 () => {
811 unitTest(
812 "an early event from the current subscription does not establish settled routing",
813 () => {
814 expect(
815 hasReadySubscriptionEvent(
816 [event],
817 [{ identity: expected.identity, createdAt: 49_000 }],
818 {
819 ...expected,
820 readyAfter: 60_000,
821 },
822 ),
823 ).toBe(false);
824 },
825 );
826
827 unitTest(
828 "an early probe delivered later cannot substitute for a fresh probe",
829 () => {
830 expect(
831 hasReadySubscriptionEvent(
832 [event],
833 [
834 { identity: expected.identity, createdAt: 49_000 },
835 { identity: "fresh-probe", createdAt: 61_000 },
836 ],
837 { ...expected, readyAfter: 60_000 },
838 ),
839 ).toBe(false);
840 },
841 );
842
843 unitTest(
844 "a fresh matching probe establishes readiness after the propagation interval",
845 () => {
846 expect(
847 hasReadySubscriptionEvent(
848 [event],
849 [{ identity: expected.identity, createdAt: 61_000 }],
850 {
851 ...expected,
852 readyAfter: 60_000,
853 },
854 ),
855 ).toBe(true);
856 },
857 );
858
859 unitTest("matches the exact Vectorize event envelope", () => {
860 const decoded = Schema.decodeUnknownSync(SubscriptionEvent)(
861 JSON.stringify(event),
862 );
863 expect(matchesSubscriptionEvent(decoded, expected)).toBe(true);
864 });
865
866 unitTest(
867 "an index name containing the subscription id does not prove subscription identity",
868 () => {
869 expect(
870 matchesSubscriptionEvent(
871 {
872 ...event,
873 metadata: { ...event.metadata, eventSubscriptionId: "other" },
874 },
875 expected,
876 ),
877 ).toBe(false);
878 },
879 );
880
881 unitTest("a readiness probe cannot satisfy a distinct final index", () => {
882 expect(
883 matchesSubscriptionEvent(event, {
884 ...expected,
885 identity: "final-index",
886 }),
887 ).toBe(false);
888 });
889
890 unitTest("matches the whole resource name, not a substring", () => {
891 expect(
892 matchesSubscriptionEvent(
893 {
894 ...event,
895 payload: { name: `${expected.identity}-other` },
896 },
897 expected,
898 ),
899 ).toBe(false);
900 });
901
902 unitTest(
903 "rejects another account even when its name contains the expected account",
904 () => {
905 expect(
906 matchesSubscriptionEvent(
907 {
908 ...event,
909 metadata: { ...event.metadata, accountId: "other" },
910 payload: { name: expected.identity, id: expected.accountId },
911 },
912 expected,
913 ),
914 ).toBe(false);
915 },
916 );
917
918 unitTest("requires the exact event type and product source", () => {
919 expect(
920 matchesSubscriptionEvent(
921 { ...event, type: `${event.type}.other` },
922 expected,
923 ),
924 ).toBe(false);
925 expect(
926 matchesSubscriptionEvent(
927 { ...event, source: { type: "r2" } },
928 expected,
929 ),
930 ).toBe(false);
931 });
932
933 unitTest(
934 "matches an Images upload by exact payload id and subscription",
935 () => {
936 const image = {
937 ...event,
938 type: "cf.images.image.uploaded",
939 source: { type: "images" },
940 payload: { id: "subscription-final" },
941 };
942 const target = {
943 ...expected,
944 source: "images" as const,
945 type: "image.uploaded",
946 identity: "subscription-final",
947 };
948 expect(matchesSubscriptionEvent(image, target)).toBe(true);
949 expect(
950 matchesSubscriptionEvent(
951 { ...image, payload: { id: "subscription-probe-0" } },
952 target,
953 ),
954 ).toBe(false);
955 expect(
956 matchesSubscriptionEvent(
957 {
958 ...image,
959 metadata: {
960 ...image.metadata,
961 eventSubscriptionId: "old-subscription",
962 },
963 },
964 target,
965 ),
966 ).toBe(false);
967 expect(
968 matchesSubscriptionEvent(
969 { ...image, payload: { name: target.identity } },
970 target,
971 ),
972 ).toBe(false);
973 },
974 );
975
976 unitTest(
977 "matches KV ids and R2 names in their documented payload fields",
978 () => {
979 expect(
980 matchesSubscriptionEvent(
981 {
982 ...event,
983 type: "cf.kv.namespace.created",
984 source: { type: "kv" },
985 payload: { id: "namespace-id", name: "namespace-name" },
986 },
987 {
988 ...expected,
989 source: "kv",
990 type: "namespace.created",
991 identity: "namespace-id",
992 },
993 ),
994 ).toBe(true);
995 expect(
996 matchesSubscriptionEvent(
997 {
998 ...event,
999 type: "cf.r2.bucket.created",
1000 source: { type: "r2" },
1001 payload: { name: "bucket-name" },
1002 },
1003 {
1004 ...expected,
1005 source: "r2",
1006 type: "bucket.created",
1007 identity: "bucket-name",
1008 },
1009 ),
1010 ).toBe(true);
1011 },
1012 );
1013 },
1014);
1015
1016// Native routing experiments are opt-in; control-plane reads are not delivery barriers.
1017for (const change of ["replacement", "update", "paused-update"] as const) {
1018 const enabled =
1019 process.env[
1020 {
1021 replacement: "CLOUDFLARE_TEST_VECTORIZE_IMMEDIATE_REPLACEMENT",
1022 update: "CLOUDFLARE_TEST_VECTORIZE_IMMEDIATE_UPDATE",
1023 "paused-update": "CLOUDFLARE_TEST_VECTORIZE_PAUSED_UPDATE",
1024 }[change]
1025 ] === "1";
1026 test.provider.skipIf(!enabled)(
1027 `Vectorize immediately routes post-${change} events to the current subscription`,
1028 (stack) =>
1029 Effect.gen(function* () {
1030 yield* stack.destroy();
1031 const { accountId } = yield* yield* CloudflareEnvironment;
1032 const start = yield* Clock.currentTimeMillis;
1033 const cleanup = makeSubscriptionCleanup();
1034 const makeQueue = (suffix: string) =>
1035 Effect.gen(function* () {
1036 const queue = yield* Effect.acquireRelease(
1037 queues.createQueue({
1038 accountId,
1039 queueName: `alchemy-vectorize-replacement-${suffix}`,
1040 }),
1041 (queue) =>
1042 queues
1043 .deleteQueue({ accountId, queueId: queue.queueId! })
1044 .pipe(cleanup),
1045 );
1046 yield* queues.createConsumer({
1047 accountId,
1048 queueId: queue.queueId!,
1049 type: "http_pull",
1050 });
1051 return queue.queueId!;
1052 });
1053 const a = yield* makeQueue("a");
1054 const b = yield* makeQueue("b");
1055 const makeSubscription = (queueId: string) =>
1056 Effect.acquireRelease(
1057 queues.createSubscription({
1058 accountId,
1059 name: `native-${queueId}`,
1060 source: { type: "vectorize" },
1061 events: ["index.created"],
1062 enabled: true,
1063 destination: { type: "queues.queue", queueId },
1064 }),
1065 (subscription) =>
1066 queues
1067 .deleteSubscription({
1068 accountId,
1069 subscriptionId: subscription.id,
1070 })
1071 .pipe(
1072 Effect.catchTag("SubscriptionNotFound", () => Effect.void),
1073 cleanup,
1074 ),
1075 );
1076 const create = (name: string) =>
1077 Effect.gen(function* () {
1078 const index = yield* Effect.acquireRelease(
1079 vectorize.createIndex({
1080 accountId,
1081 name,
1082 config: { dimensions: 32, metric: "cosine" },
1083 }),
1084 () =>
1085 vectorize
1086 .deleteIndex({ accountId, indexName: name })
1087 .pipe(cleanup),
1088 );
1089 yield* Effect.logInfo(`Native ${change} create`, {
1090 elapsedMs: (yield* Clock.currentTimeMillis) - start,
1091 index,
1092 });
1093 return name;
1094 });
1095 const received: { queueId: string; event: SubscriptionEvent }[] = [];
1096 const pull = (queueId: string) =>
1097 Effect.gen(function* () {
1098 const batch = yield* queues
1099 .pullMessage({
1100 accountId,
1101 queueId,
1102 batchSize: 100,
1103 visibilityTimeoutMs: 30_000,
1104 })
1105 .pipe(
1106 Effect.retry({
1107 while: (error) => error._tag === "QueueHttpPullNotEnabled",
1108 schedule: Schedule.spaced("2 seconds"),
1109 times: 8,
1110 }),
1111 );
1112 for (const message of batch.messages ?? []) {
1113 if (message.body) {
1114 const event = yield* Schema.decodeUnknownEffect(
1115 SubscriptionEvent,
1116 )(message.body);
1117 expect(event.type).toBe("cf.vectorize.index.created");
1118 expect(event.source.type).toBe("vectorize");
1119 expect(event.metadata.accountId).toBe(accountId);
1120 received.push({ queueId, event });
1121 yield* Effect.logInfo(`Native ${change} receipt`, {
1122 elapsedMs: (yield* Clock.currentTimeMillis) - start,
1123 queueId,
1124 messageId: message.id,
1125 timestampMs: message.timestampMs,
1126 attempts: message.attempts,
1127 rawBody: message.body,
1128 event,
1129 });
1130 }
1131 }
1132 const acks = (batch.messages ?? []).flatMap(({ leaseId }) =>
1133 leaseId ? [{ leaseId }] : [],
1134 );
1135 if (acks.length) {
1136 const ack = yield* queues.ackMessage({
1137 accountId,
1138 queueId,
1139 acks,
1140 });
1141 expect(ack.ackCount).toBe(acks.length);
1142 expect(Object.keys(ack.warnings ?? {})).toHaveLength(0);
1143 }
1144 });
1145 const old = yield* makeSubscription(a);
1146 yield* pull(b);
1147 const initialProbes: string[] = [];
1148 const initiallyReady = () =>
1149 received.some(
1150 ({ queueId, event }) =>
1151 queueId === a &&
1152 event.metadata.eventSubscriptionId === old.id &&
1153 initialProbes.includes(event.payload.name!),
1154 );
1155 yield* Effect.gen(function* () {
1156 initialProbes.push(yield* create(`vec-${a}-${initialProbes.length}`));
1157 yield* pull(a);
1158 }).pipe(
1159 Effect.repeat({
1160 schedule: Schedule.spaced("7 seconds"),
1161 times: 8,
1162 until: initiallyReady,
1163 }),
1164 );
1165 expect(initiallyReady()).toBe(true);
1166 const observe = Effect.fn(function* (
1167 subscriptionId: string,
1168 phase: string,
1169 ) {
1170 const observed = yield* queues.getSubscription({
1171 accountId,
1172 subscriptionId,
1173 });
1174 yield* Effect.logInfo(`Native ${change} configuration`, {
1175 phase,
1176 elapsedMs: (yield* Clock.currentTimeMillis) - start,
1177 subscription: observed,
1178 });
1179 return observed;
1180 });
1181 const current = yield* Effect.gen(function* () {
1182 if (change === "paused-update") {
1183 yield* queues.patchSubscription({
1184 accountId,
1185 subscriptionId: old.id,
1186 enabled: false,
1187 });
1188 const paused = yield* observe(old.id, "disabled");
1189 expect(paused.enabled).toBe(false);
1190 expect(paused.destination.queueId).toBe(a);
1191 yield* create(`vec-${a}-disabled`);
1192 yield* pull(a);
1193 yield* pull(b);
1194 yield* queues.patchSubscription({
1195 accountId,
1196 subscriptionId: old.id,
1197 destination: { type: "queues.queue", queueId: b },
1198 });
1199 const moved = yield* observe(
1200 old.id,
1201 "destination-updated-while-disabled",
1202 );
1203 expect(moved.enabled).toBe(false);
1204 expect(moved.destination.queueId).toBe(b);
1205 const resumed = yield* queues.patchSubscription({
1206 accountId,
1207 subscriptionId: old.id,
1208 enabled: true,
1209 });
1210 expect(resumed.id).toBe(old.id);
1211 return resumed;
1212 }
1213 if (change === "update") {
1214 const updated = yield* queues.patchSubscription({
1215 accountId,
1216 subscriptionId: old.id,
1217 destination: { type: "queues.queue", queueId: b },
1218 });
1219 expect(updated.id).toBe(old.id);
1220 return updated;
1221 }
1222 yield* queues.deleteSubscription({
1223 accountId,
1224 subscriptionId: old.id,
1225 });
1226 yield* queues
1227 .getSubscription({ accountId, subscriptionId: old.id })
1228 .pipe(
1229 Effect.flatMap(() =>
1230 Effect.fail(new Error("Deleted subscription still exists")),
1231 ),
1232 Effect.catchTag("SubscriptionNotFound", () => Effect.void),
1233 );
1234 return yield* makeSubscription(b);
1235 });
1236 const observed = yield* observe(
1237 current.id,
1238 "active-at-new-destination",
1239 );
1240 expect(observed.destination.queueId).toBe(b);
1241 expect(observed.enabled).toBe(true);
1242 yield* Effect.logInfo(`Native ${change} identities`, {
1243 oldSubscription: old.id,
1244 oldQueue: a,
1245 newSubscription: current.id,
1246 newQueue: b,
1247 elapsedMs: (yield* Clock.currentTimeMillis) - start,
1248 });
1249 const identities = yield* Effect.forEach([0, 1, 2], (i) =>
1250 create(`vec-${b}-after-${change}-${i}`),
1251 );
1252 yield* Effect.gen(function* () {
1253 yield* pull(a);
1254 yield* pull(b);
1255 }).pipe(
1256 Effect.repeat({
1257 schedule: Schedule.spaced("2 seconds"),
1258 times: 5,
1259 }),
1260 );
1261 const deliveries = received.filter(({ event }) =>
1262 identities.includes(event.payload.name!),
1263 );
1264 yield* Effect.logInfo(`Native ${change} result`, {
1265 identities,
1266 deliveries,
1267 });
1268 expect(deliveries.filter(({ queueId }) => queueId === a)).toEqual([]);
1269 for (const identity of identities) {
1270 expect(
1271 deliveries.some(
1272 ({ queueId, event }) =>
1273 queueId === b &&
1274 event.metadata.eventSubscriptionId === current.id &&
1275 event.payload.name === identity,
1276 ),
1277 ).toBe(true);
1278 }
1279 yield* stack.destroy();
1280 }).pipe(
1281 Effect.timeout("90 seconds"),
1282 Effect.scoped,
1283 Effect.ensuring(
1284 stack
1285 .destroy()
1286 .pipe(
1287 Effect.timeout("5 seconds"),
1288 Effect.orDie,
1289 Effect.interruptible,
1290 ),
1291 ),
1292 ),
1293 {
1294 tags: [
1295 "provider:cloudflare",
1296 "provider:cloudflare:queue",
1297 "provider:cloudflare:vectorize",
1298 "live",
1299 ],
1300 timeout: 120_000,
1301 exclusive: true,
1302 },
1303 );
1304}