[pull] master from apify:master

packages/core/src/storages/sitemap_request_list.ts

@@ -505,18 +505,26 @@ export class SitemapRequestList implements IRequestList {
 
         // Create a new stream, as we have read all the URLs from the current one.
         // Pushing the urls back to the original stream might not be possible if it has been ended.
-        const newStream = this.createNewStream(this.urlQueueStream.readableHighWaterMark);
+        const previousStream = this.urlQueueStream;
+        const newStream = this.createNewStream(previousStream.readableHighWaterMark);
 
         for (const url of urlQueue) {
             newStream.push(url);
         }
 
-        if (this.urlQueueStream.writableEnded) {
+        if (previousStream.writableEnded) {
             newStream.end();
         }
 
         this.urlQueueStream = newStream;
 
+        // A `pushNextUrl()` call may be blocked on backpressure, waiting for a `readdata` event on the
+        // previous stream. That event is only ever emitted by `readNextUrl()` on the current stream, so
+        // after the swap the waiter would never be notified and the background sitemap loading would hang.
+        // Re-emit `readdata` on the previous stream to release any such pending waiter (its URL has already
+        // been transferred to the new stream above).
+        previousStream.emit('readdata');
+
         await this.store.setValue(this.persistStateKey, {
             sitemapParsingProgress: {
                 pendingSitemapUrls: Array.from(this.sitemapParsingProgress.pendingSitemapUrls),

test/core/sitemap_request_list.test.ts

@@ -690,4 +690,41 @@ describe('SitemapRequestList', () => {
         // The strategy is persisted on the request so it keeps being enforced after navigation.
         expect((request as any).enqueueStrategy).toBe('same-hostname');
     });
+
+    test('persistState does not deadlock a backpressured sitemap load', async () => {
+        // `maxBufferSize: 1` makes the background loader block on backpressure right after
+        // pushing the first URL. Persisting the state at that moment swaps the underlying stream,
+        // which used to orphan the pending push and hang the loading indefinitely.
+        const list = await SitemapRequestList.open({
+            sitemapUrls: [`${url}/sitemap.xml`],
+            persistStateKey: 'backpressure-persist',
+            maxBufferSize: 1,
+            enqueueStrategy: 'all',
+        });
+
+        // Wait until the first URL is buffered, i.e. the loader is parked on backpressure.
+        while (await list.isEmpty()) {
+            await sleep(20);
+        }
+
+        await list.persistState();
+
+        const urls = new Set<string>();
+        for await (const request of list) {
+            await list.markRequestHandled(request);
+            urls.add(request.url);
+        }
+
+        expect(list.isSitemapFullyLoaded()).toBe(true);
+        await expect(list.isFinished()).resolves.toBe(true);
+        expect(urls).toEqual(
+            new Set([
+                'http://not-exists.com/',
+                'http://not-exists.com/catalog?item=12&desc=vacation_hawaii',
+                'http://not-exists.com/catalog?item=73&desc=vacation_new_zealand',
+                'http://not-exists.com/catalog?item=74&desc=vacation_newfoundland',
+                'http://not-exists.com/catalog?item=83&desc=vacation_usa',
+            ]),
+        );
+    }, 10_000);
 });