diff --git a/docs/guides/result_storage.mdx b/docs/guides/result_storage.mdx index cb7103ce52c9..67aa834162d0 100644 --- a/docs/guides/result_storage.mdx +++ b/docs/guides/result_storage.mdx @@ -148,6 +148,30 @@ This is **on by default** for every crawler and covers the whole request lifecyc - **Deferred cleanups.** Callbacks registered with `registerDeferredCleanup()` run after the transaction is closed, so their writes land immediately and are not rolled back. In `AdaptivePlaywrightCrawler` they run once per request handler attempt, so a write there can land twice for one request — push your results from the request handler instead. +### Reacting to a commit + +A write made inside a transaction cannot fail where it is made — `pushData()` only records it, and the failure, if any, happens at commit time. `afterStorageCommit()` is where you get it back: the callback runs once the request's writes have been committed, or once committing them has failed, and receives the error in the latter case. + +```typescript +async function requestHandler({ pushData, afterStorageCommit, useState }) { + const state = await useState({ itemCount: 0 }); + + await pushData(item); + afterStorageCommit((error) => { + if (error) { + throw new NonRetryableError('The page is too big to store', { cause: error }); + } + state.itemCount++; + }); +} +``` + +Both halves of that callback need the commit. Incrementing the counter in the handler would over-count, because `useState()` is not transactional: a request that pushes an item and then fails keeps the increment but loses the item. And a commit failure left alone is just an ordinary request error — retried to exhaustion, with nothing tying it to the write that caused it. + +Callbacks run in registration order, after the commit and before the request is marked as handled. The first one to throw propagates and the rest do not run; on a failed commit its error replaces the commit error, which is how the example above decides what the request fails with. After a *successful* commit, a callback that throws fails the request **without a retry** whatever it throws — the writes are already durable and retrying would duplicate them, so anything that is not already a `NonRetryableError` is wrapped in an `AfterCommitError`. Callbacks are *not* run when the request handler itself fails — nothing was written, and that is what `errorHandler` / `failedRequestHandler` are for. + +In `AdaptivePlaywrightCrawler` a callback belongs to the request handler attempt that registered it, so it runs only for the attempt whose writes are the ones being committed. Under `transactionalStorage: false` there is no commit to react to — writes are applied as they are made and throw at the call site on their own — and `afterStorageCommit()` throws. + ### Escape hatches The feature is escapable at three granularities: diff --git a/docs/public-api/crawlee-core.api.md b/docs/public-api/crawlee-core.api.md index aab338a54dee..fa7714a1d168 100644 --- a/docs/public-api/crawlee-core.api.md +++ b/docs/public-api/crawlee-core.api.md @@ -65,6 +65,11 @@ export interface AddRequestsBatchedResult { waitForAllRequestsToBeAdded: Promise; } +// @public +export class AfterCommitError extends NonRetryableError { + constructor(cause: unknown); +} + // @public export class ApifyLogAdapter extends BaseCrawleeLogger { constructor(apifyLog: Log, options?: Partial); @@ -300,6 +305,7 @@ export { CrawleeLoggerOptions } // @public (undocumented) export interface CrawlingContext extends RestrictedCrawlingContext { + afterStorageCommit(callback: (error?: Error) => Awaitable): void; extendTimeout(secs: number): void; registerDeferredCleanup(cleanup: () => Promise): void; sendRequest: (requestOverrides?: Partial, optionsOverrides?: SendRequestOptions) => Promise; @@ -2005,6 +2011,7 @@ export class StorageStatsTracker> { // @public export class StorageTransaction implements StorageTransactionView { + afterCommit(callback: (error?: Error) => Awaitable): void; commit(): Promise; // (undocumented) get datasetItems(): { diff --git a/docs/upgrading/upgrading_v4.md b/docs/upgrading/upgrading_v4.md index e151e40c28ec..aa151f8cc34c 100644 --- a/docs/upgrading/upgrading_v4.md +++ b/docs/upgrading/upgrading_v4.md @@ -671,9 +671,10 @@ Every crawler now wraps each request in a **storage transaction** (see the [Tran The observable behavior of a *successful* handler is unchanged (reads within a handler see its own writes), but several things differ on the failure path and around handler boundaries: - **Uncommitted writes are invisible to other handlers.** Using the key-value store as a live channel between concurrently running handlers no longer works — one handler's `setValue()` only becomes visible to others once its request succeeds. Use `useState()` for cross-handler communication. -- **`useState()` / `getAutoSavedValue()` are *not* transactional.** The shared state object stays live; mutations of it are not rolled back when a handler fails. +- **`useState()` / `getAutoSavedValue()` are *not* transactional.** The shared state object stays live; mutations of it are not rolled back when a handler fails. Side effects that have to match the writes that actually landed — result counters above all — belong in a callback registered with the new `afterStorageCommit()` context helper. - **Request queue additions are applied immediately by default** (the `writeThrough` policy) and are not rolled back — deduplication by `uniqueKey` keeps retries idempotent. Pass `transactionalStorage: { requestQueue: 'deferred' }` for strict all-or-nothing enqueues. - **Commit is at-least-once.** It spans multiple storages, so a commit that fails partway fails the request; the retry may re-apply writes that already landed. +- **A write cannot fail where it is made.** `pushData()` records the item and returns; if the storage backend rejects it, that happens at commit time, and a `try`/`catch` around the call never sees it. Register an `afterStorageCommit()` callback next to the write instead — it receives the commit error, and an error it throws replaces it, so a rejected write can still be turned into a `NonRetryableError` of your own. - **`KeyValueStore.setValue()` with a stream value throws inside a request handler.** A stream can only be consumed once, so it cannot be buffered. Wrap the call in `withDirectStorageAccess()` to write it immediately: ```typescript @@ -690,7 +691,7 @@ The mechanism can be disabled entirely with `transactionalStorage: false` on any #### Removed symbols and options - `checkStorageAccess` and `withCheckedStorageAccess` are superseded by the transaction mechanism; the per-call-site helper is now `withDirectStorageAccess()`. -- The experimental `AdaptivePlaywrightCrawler` no longer needs its bespoke write-buffering machinery: the `preventDirectStorageAccess` option is gone (direct storage calls are now captured by the per-attempt transaction instead of throwing), and `RequestHandlerResult` is replaced by the read-only `StorageTransactionView`, which the `resultChecker` / `resultComparator` callbacks (and `fullResultComparator`) now receive. The view keeps the familiar accessors (`datasetItems`, `enqueuedUrls`, `keyValueStoreChanges`), so most callbacks only need a type change. The `calls` and `enqueuedUrlLists` accessors are gone — `requestsFromUrl` sources are now expanded when added, so the fetched URLs appear in `enqueuedUrls` (and are what `fullResultComparator` compares). +- The experimental `AdaptivePlaywrightCrawler` no longer needs its bespoke write-buffering machinery: the `preventDirectStorageAccess` option is gone (direct storage calls are now captured by the per-attempt transaction instead of throwing), and `RequestHandlerResult` is replaced by the read-only `StorageTransactionView`, which the `resultChecker` / `resultComparator` callbacks (and `fullResultComparator`) now receive. The view keeps the familiar accessors (`datasetItems`, `enqueuedUrls`, `keyValueStoreChanges`), so most callbacks only need a type change. The `calls` and `enqueuedUrlLists` accessors are gone — `requestsFromUrl` sources are now expanded when added, so the fetched URLs appear in `enqueuedUrls` (and are what `fullResultComparator` compares). The `commitResult` override point is gone too; use an `afterStorageCommit()` callback to run logic once the winning attempt's writes have landed (or failed to). ### `storageObject` is removed from storage classes diff --git a/packages/basic-crawler/src/internals/basic-crawler.ts b/packages/basic-crawler/src/internals/basic-crawler.ts index 8c5bdfd3ae2c..b3501d37ed99 100644 --- a/packages/basic-crawler/src/internals/basic-crawler.ts +++ b/packages/basic-crawler/src/internals/basic-crawler.ts @@ -1472,6 +1472,19 @@ export class BasicCrawler< registerDeferredCleanup: (cleanup: () => Promise) => { deferredCleanup.push(cleanup); }, + afterStorageCommit: (callback: (error?: Error) => Awaitable) => { + const transaction = currentStorageTransaction(); + + if (transaction === undefined) { + throw new Error( + 'afterStorageCommit() needs an active storage transaction, and there is none. ' + + "With `transactionalStorage: false` the request's writes are applied as they are " + + 'made, so they throw at the call site instead of at commit time.', + ); + } + + transaction.afterCommit(callback); + }, extendTimeout: (secs: number) => { const extraMillis = secs * 1000; diff --git a/packages/core/src/crawlers/crawler_commons.ts b/packages/core/src/crawlers/crawler_commons.ts index 0420c81ebdf4..87dc360fc647 100644 --- a/packages/core/src/crawlers/crawler_commons.ts +++ b/packages/core/src/crawlers/crawler_commons.ts @@ -1,4 +1,11 @@ -import type { Dictionary, HttpRequestOptions, ISession, ProxyInfo, SendRequestOptions } from '@crawlee/types'; +import type { + Awaitable, + Dictionary, + HttpRequestOptions, + ISession, + ProxyInfo, + SendRequestOptions, +} from '@crawlee/types'; import type { ReadonlyDeep } from 'type-fest'; import type { EnqueueUrlsOptions } from '../enqueue_links/enqueue_links.js'; @@ -183,6 +190,47 @@ export interface CrawlingContext exten */ registerDeferredCleanup(cleanup: () => Promise): void; + /** + * Registers `callback` to run once the request's storage writes have been committed, or once + * committing them has failed - the point where a write made in the handler can finally be reacted + * to where it was made. Two things belong here: side effects that must agree with what actually + * landed (result counters, progress reporting), and handling of a write that did not land at all. + * + * ```ts + * async requestHandler({ pushData, afterStorageCommit, useState }) { + * const state = await useState({ itemCount: 0 }); + * + * await pushData(item); + * afterStorageCommit((error) => { + * if (error) { + * throw new NonRetryableError('The page is too big to store', { cause: error }); + * } + * state.itemCount++; + * }); + * } + * ``` + * + * `useState()` is deliberately not transactional, which is what makes the counter case necessary: + * incrementing in the handler would keep the increment for a request that pushes an item and then + * fails, while the item itself is rolled back. + * + * Callbacks run in registration order, after the commit and before the request is marked as handled. + * The first one to throw propagates and the rest do not run; on a failed commit its error replaces + * the commit error, so a callback can decide what the request fails with. After a *successful* + * commit, a callback that throws fails the request **without a retry** whatever it throws - the + * writes are already durable and retrying would duplicate them, so anything that is not already a + * {@apilink NonRetryableError} is wrapped in an {@apilink AfterCommitError}. + * + * Callbacks are *not* run when the request handler itself fails - nothing was written, and + * `errorHandler` / `failedRequestHandler` cover that. + * + * Throws when storage is not transactional (`transactionalStorage: false`) - writes are then applied + * as they are made and throw at the call site on their own. In {@apilink AdaptivePlaywrightCrawler} + * the callback belongs to the current request handler attempt, so it runs only for the attempt whose + * writes are the ones being committed. + */ + afterStorageCommit(callback: (error?: Error) => Awaitable): void; + /** * Gives the current request `secs` more seconds to finish, for when how long it needs is only apparent * once it is already running - a listing page that turns out to have far more to scroll through than diff --git a/packages/core/src/errors.ts b/packages/core/src/errors.ts index f558e86b5eee..c0dc7d6b73dd 100644 --- a/packages/core/src/errors.ts +++ b/packages/core/src/errors.ts @@ -145,6 +145,19 @@ export class RequestHandlerError extends Error { } } +/** + * Wraps the failure of a callback registered with {@apilink StorageTransaction.afterCommit|`afterCommit`} + * that ran after the request's writes had already been committed. + * + * Non-retryable by nature: the writes are durable, so re-running the request handler would duplicate + * them. A callback that throws a `NonRetryableError` of its own is left alone. + */ +export class AfterCommitError extends NonRetryableError { + constructor(cause: unknown) { + super(cause instanceof Error ? cause.message : String(cause), { cause }); + } +} + /** * Thrown when attempting to set a different service instance after one has already been retrieved. */ diff --git a/packages/core/src/storages/transaction.ts b/packages/core/src/storages/transaction.ts index b16abd1b9cb1..ca6771c45b15 100644 --- a/packages/core/src/storages/transaction.ts +++ b/packages/core/src/storages/transaction.ts @@ -5,6 +5,7 @@ import type { Awaitable, Dictionary } from '@crawlee/types'; import { addTimeoutToPromise, storage as timeoutStorage, tryCancel } from '@apify/timeout'; import { serviceLocator } from '../service_locator.js'; +import { AfterCommitError, NonRetryableError } from '../errors.js'; import type { RecordOptions } from './key_value_store.js'; /** @@ -175,6 +176,9 @@ export class StorageTransaction implements StorageTransactionView { #disposed = false; + /** Callbacks registered via `afterCommit()`, run in registration order once a commit has been attempted. */ + readonly #afterCommitCallbacks: ((error?: Error) => Awaitable)[] = []; + /** @internal */ constructor(options: StorageTransactionOptions = {}) { this.policy = { ...DEFAULT_STORAGE_WRITE_POLICY, ...options.policy }; @@ -210,6 +214,20 @@ export class StorageTransaction implements StorageTransactionView { this.journal.push(entry); } + /** + * Registers a callback to run once a commit has been attempted, whether it succeeded (`error` is + * `undefined`) or failed - the only point at which a deferred write can be reacted to where it was + * made. Callbacks run in registration order; the first to throw propagates, the rest do not run, and + * on a failed commit its error replaces the commit error. Not run on rollback: nothing was written. + */ + afterCommit(callback: (error?: Error) => Awaitable): void { + if (!this.isActive) { + throw new Error(`Cannot register a commit callback on a transaction in the '${this.#state}' state`); + } + + this.#afterCommitCallbacks.push(callback); + } + /** * Replays the journaled writes into real storage. A no-op unless the transaction is `open`. * @@ -237,11 +255,33 @@ export class StorageTransaction implements StorageTransactionView { `Committing the storage transaction timed out after ${this.#commitTimeoutMillis / 1000} seconds.`, ), ); - this.#state = 'committed'; } catch (error) { this.#state = 'failed'; - throw error; + // Callbacks are handed an `Error`, so a backend that rejects with anything else gets wrapped. + const commitError = error instanceof Error ? error : new Error(String(error), { cause: error }); + + // A callback that throws here propagates instead: it was handed the failure and decided what + // the request should fail with. + await this.#runAfterCommitCallbacks(commitError); + throw commitError; } + + this.#state = 'committed'; + + try { + await this.#runAfterCommitCallbacks(); + } catch (error) { + throw error instanceof NonRetryableError ? error : new AfterCommitError(error); + } + } + + /** Fresh timeout context for the same reason as the flush. */ + async #runAfterCommitCallbacks(error?: Error): Promise { + await timeoutStorage.exit(async () => { + for (const callback of this.#afterCommitCallbacks) { + await callback(error); + } + }); } private async flush(): Promise { @@ -282,9 +322,10 @@ export class StorageTransaction implements StorageTransactionView { } /** - * Releases the journal and the write-time snapshots it holds. Must be called for *every* terminal - * state, `failed` included. Idempotent, never throws, and does not change `state`. Any - * {@apilink StorageTransactionView} of this transaction is only valid until this is called. + * Releases the journal, the write-time snapshots it holds and any registered commit callbacks. Must + * be called for *every* terminal state, `failed` included. Idempotent, never throws, and does not + * change `state`. Any {@apilink StorageTransactionView} of this transaction is only valid until this + * is called. */ dispose(): void { if (this.#disposed) { @@ -305,6 +346,7 @@ export class StorageTransaction implements StorageTransactionView { this.#disposed = true; this.journal.length = 0; + this.#afterCommitCallbacks.length = 0; } get datasetItems(): { item: Dictionary; datasetId: string }[] { diff --git a/test/core/crawlers/adaptive_playwright_crawler.test.ts b/test/core/crawlers/adaptive_playwright_crawler.test.ts index d5cd9967743f..c4f72c0ab4b8 100644 --- a/test/core/crawlers/adaptive_playwright_crawler.test.ts +++ b/test/core/crawlers/adaptive_playwright_crawler.test.ts @@ -834,6 +834,81 @@ describe('AdaptivePlaywrightCrawler', () => { expect(await store.getValue('1')).toEqual({ content: 42 }); }); + test('should run afterStorageCommit callbacks only for the committed attempt', async () => { + // Always detect: the browser attempt is committed, then a static attempt runs purely for the + // comparison and is discarded. So the handler runs twice for one request. + const renderingTypePredictor = makeRiggedRenderingTypePredictor({ + detectionProbabilityRecommendation: 1, + renderingType: 'clientOnly', + }); + + let handlerRuns = 0; + const committed: number[] = []; + + const requestHandler: AdaptivePlaywrightCrawlerOptions['requestHandler'] = async ({ + pushData, + afterStorageCommit, + }) => { + const run = ++handlerRuns; + await pushData({ run }); + afterStorageCommit(() => void committed.push(run)); + }; + + const crawler = await makeOneshotCrawler( + { + requestHandler, + renderingTypePredictor, + maxRequestsPerCrawl: 1, + maxRequestRetries: 0, + }, + [`http://${HOSTNAME}:${port}/static`], + ); + + await crawler.run(); + + expect(handlerRuns).toBe(2); + expect(committed).toEqual([1]); + expect((await Dataset.getData()).items).toEqual([{ run: 1 }]); + }); + + test('should not retry a request whose afterStorageCommit callback threw after a commit', async () => { + const renderingTypePredictor = makeRiggedRenderingTypePredictor({ + detectionProbabilityRecommendation: 0, + renderingType: 'static', + }); + + let handlerRuns = 0; + const failedRequestHandler = vi.fn(); + + const requestHandler: AdaptivePlaywrightCrawlerOptions['requestHandler'] = async ({ + pushData, + afterStorageCommit, + }) => { + handlerRuns++; + await pushData({ run: handlerRuns }); + afterStorageCommit(() => { + throw new Error('bookkeeping failed'); + }); + }; + + const crawler = await makeOneshotCrawler( + { + requestHandler, + renderingTypePredictor, + maxRequestsPerCrawl: 1, + maxRequestRetries: 3, + failedRequestHandler, + }, + [`http://${HOSTNAME}:${port}/static`], + ); + + await crawler.run(); + + expect(handlerRuns).toBe(1); + expect(failedRequestHandler).toHaveBeenCalledTimes(1); + expect((await Dataset.getData()).items).toEqual([{ run: 1 }]); + }); + test('should persist RenderingTypePredictor state on PERSIST_STATE events', async () => { const requestHandler: AdaptivePlaywrightCrawlerOptions['requestHandler'] = vi.fn(async ({ pushData }) => { await pushData({ content: 'test data' }); diff --git a/test/core/crawlers/basic_crawler.test.ts b/test/core/crawlers/basic_crawler.test.ts index 86f1665c6a1c..34250905e1f0 100644 --- a/test/core/crawlers/basic_crawler.test.ts +++ b/test/core/crawlers/basic_crawler.test.ts @@ -13,6 +13,7 @@ import type { } from '@crawlee/basic'; import type { Session } from '@crawlee/basic'; import { + AfterCommitError, BasicCrawler, Configuration, CriticalError, @@ -3580,6 +3581,113 @@ describe('BasicCrawler', () => { await expect(dataset.getData()).resolves.toMatchObject({ items: [{ from: 'failing-handler' }] }); }); + test('afterStorageCommit runs for the attempt whose writes were committed', async () => { + const committedAttempts: number[] = []; + + const crawler = new BasicCrawler({ + maxRequestRetries: 1, + requestHandler: async ({ request, pushData, afterStorageCommit }) => { + await pushData({ attempt: request.retryCount }); + afterStorageCommit(() => void committedAttempts.push(request.retryCount)); + + if (request.retryCount === 0) { + throw new Error('first attempt fails'); + } + }, + }); + + await crawler.run([`http://${HOSTNAME}:${port}/`]); + + // The rolled-back attempt registered a callback as well; only the committed one ran it. + expect(committedAttempts).toEqual([1]); + }); + + test('a callback that throws after a successful commit does not retry the request', async () => { + const failures: Error[] = []; + const retried: Error[] = []; + let handlerRuns = 0; + + const crawler = new BasicCrawler({ + maxRequestRetries: 3, + requestHandler: async ({ pushData, afterStorageCommit }) => { + handlerRuns++; + await pushData({ item: true }); + // A plain, ordinarily retryable error: the retry is suppressed because the item is + // already committed and re-running the handler would push it a second time. + afterStorageCommit(() => { + throw new Error('bookkeeping failed'); + }); + }, + errorHandler: async (_context, error) => { + retried.push(error); + }, + failedRequestHandler: async (_context, error) => { + failures.push(error); + }, + }); + + await crawler.run([`http://${HOSTNAME}:${port}/`]); + + // Reaches `failedRequestHandler` with the wrapper intact, so a handler can tell that the + // items did land. `errorHandler` only runs for retried requests, so it is skipped. + expect(failures).toEqual([expect.any(AfterCommitError)]); + expect(failures[0].cause).toMatchObject({ message: 'bookkeeping failed' }); + expect(retried).toEqual([]); + expect(handlerRuns).toBe(1); + await expect(Dataset.getData()).resolves.toMatchObject({ total: 1 }); + }); + + test('afterStorageCommit turns a rejected write into a non-retryable request failure', async () => { + const dataset = await Dataset.open(); + vitest + .spyOn(dataset.backend, 'pushData') + .mockRejectedValue(new Error('Data item is too large (size: 10000000 bytes)')); + + const failures: string[] = []; + let handlerRuns = 0; + + const crawler = new BasicCrawler({ + maxRequestRetries: 3, + requestHandler: async ({ pushData, afterStorageCommit }) => { + handlerRuns++; + await pushData({ huge: true }); + afterStorageCommit((error) => { + if (error?.message.includes('too large')) { + throw new NonRetryableError('Enable `saveHtmlAsFile`', { cause: error }); + } + }); + }, + failedRequestHandler: async (_context, error) => { + failures.push(error.message); + }, + }); + + await crawler.run([`http://${HOSTNAME}:${port}/`]); + + // Left alone, the commit failure is an ordinary request error and gets retried to exhaustion. + expect(failures).toEqual(['Enable `saveHtmlAsFile`']); + expect(handlerRuns).toBe(1); + }); + + test('afterStorageCommit throws when transactional storage is disabled', async () => { + const errors: string[] = []; + + const crawler = new BasicCrawler({ + maxRequestRetries: 0, + transactionalStorage: false, + requestHandler: async ({ afterStorageCommit }) => { + afterStorageCommit(() => {}); + }, + failedRequestHandler: async (_context, error) => { + errors.push(error.message); + }, + }); + + await crawler.run([`http://${HOSTNAME}:${port}/`]); + + expect(errors).toEqual([expect.stringMatching(/needs an active storage transaction/)]); + }); + test('an unclosed transaction on a normal pipeline return is discarded and logged', async () => { const crawler = new BasicCrawler({ requestHandler: async () => {} }); const errorSpy = vitest.spyOn((crawler as any).log, 'error').mockImplementation(() => {}); diff --git a/test/core/storages/storage_transaction.test.ts b/test/core/storages/storage_transaction.test.ts index 945781575ee9..a9547a5eb698 100644 --- a/test/core/storages/storage_transaction.test.ts +++ b/test/core/storages/storage_transaction.test.ts @@ -1,11 +1,13 @@ import { Readable } from 'node:stream'; import { + AfterCommitError, createStorageTransaction, Dataset, getRequestId, KeyValueStore, MemoryStorageBackend, + NonRetryableError, Request, RequestQueue, serviceLocator, @@ -110,6 +112,121 @@ describe('StorageTransaction', () => { }); }); + describe('commit callbacks', () => { + test('are not run when the transaction is rolled back', async () => { + const callback = vitest.fn(); + + const transaction = createStorageTransaction(); + await transaction.run(() => transaction.afterCommit(callback)); + transaction.rollback(); + transaction.dispose(); + + expect(callback).not.toHaveBeenCalled(); + }); + + test('receive the error of a failed commit, which still propagates', async () => { + const dataset = await Dataset.open(); + vitest.spyOn(dataset.backend, 'pushData').mockRejectedValueOnce(new Error('backend exploded')); + const callback = vitest.fn(); + + const transaction = createStorageTransaction(); + await transaction.run(async () => { + await dataset.pushData({ a: 1 }); + transaction.afterCommit(callback); + }); + + await expect(transaction.commit()).rejects.toThrow('backend exploded'); + transaction.dispose(); + + expect(callback).toHaveBeenCalledWith(expect.objectContaining({ message: 'backend exploded' })); + }); + + test('an error thrown by a callback replaces the commit error', async () => { + const dataset = await Dataset.open(); + vitest.spyOn(dataset.backend, 'pushData').mockRejectedValueOnce(new Error('Data item is too large')); + + const transaction = createStorageTransaction(); + await transaction.run(async () => { + await dataset.pushData({ a: 1 }); + transaction.afterCommit((error) => { + if (error?.message.includes('too large')) { + throw new NonRetryableError('trim the item', { cause: error }); + } + }); + }); + + await expect(transaction.commit()).rejects.toThrow(NonRetryableError); + transaction.dispose(); + }); + + test('a throwing callback fails a successful commit non-retryably', async () => { + const store = await KeyValueStore.open(); + const laterCallback = vitest.fn(); + + const transaction = createStorageTransaction(); + await transaction.run(async () => { + await store.setValue('key', { a: 1 }); + transaction.afterCommit(() => { + throw new Error('callback exploded'); + }); + transaction.afterCommit(laterCallback); + }); + + // The write is durable, so the crawler must not retry the request and duplicate it. + const error = await transaction.commit().catch((thrown) => thrown); + expect(error).toBeInstanceOf(AfterCommitError); + expect(error).toMatchObject({ message: 'callback exploded', cause: { message: 'callback exploded' } }); + + expect(transaction.state).toBe('committed'); + expect(laterCallback).not.toHaveBeenCalled(); + await expect(store.getValue('key')).resolves.toEqual({ a: 1 }); + + transaction.dispose(); + }); + + test('a callback that raises a non-retryable error of its own is left unwrapped', async () => { + const transaction = createStorageTransaction(); + const raised = new NonRetryableError('give up'); + await transaction.run(() => + transaction.afterCommit(() => { + throw raised; + }), + ); + + await expect(transaction.commit()).rejects.toBe(raised); + + transaction.dispose(); + }); + + test('run even when the ambient cancellation context has already been aborted', async () => { + const store = await KeyValueStore.open(); + + const transaction = createStorageTransaction(); + await transaction.run(() => + transaction.afterCommit(async () => store.setValue('from-callback', { ok: true })), + ); + + const controller = new AbortController(); + controller.abort(); + + // As for the flush itself: a request-handler timeout that fired before the commit must not + // cancel the storage operations of a handler that succeeded. + await timeoutStorage.run({ cancelTask: controller }, async () => transaction.commit()); + transaction.dispose(); + + await expect(store.getValue('from-callback')).resolves.toEqual({ ok: true }); + }); + + test('registering on a closed transaction throws', async () => { + const transaction = createStorageTransaction(); + await transaction.commit(); + + expect(() => transaction.afterCommit(() => {})).toThrow(/'committed' state/); + + transaction.dispose(); + }); + }); + describe('scoping helpers', () => { test('withStorageTransaction commits on success and rolls back on throw', async () => { const store = await KeyValueStore.open();