nebius/SWE-rebench-V2-PRs huggingface dataset:
Repository: dcs4cop/xcube
Pull request #154 (2019-08-14 17:14:16)

fixed memory problem with new mask_dataset_by_geometry()
Effectively only one line of code has changed, but that makes a big difference. 

Using method `dataset_copy = dataset.copy(data=<dict>)` will create a shallow `dataset` copy, but then it will literally *copy* all data from `data` into variables of `dataset_copy`, hence turning Dask arrays in `data` into numpy arrays in `dataset_copy`. Bang!   

**Last one to accept this PR should also merge immediately**


Patch:
diff --git a/xcube/util/geom.py b/xcube/util/geom.py
index c7c9b511..48038512 100644
--- a/xcube/util/geom.py
+++ b/xcube/util/geom.py
@@ -93,7 +93,7 @@ def mask_dataset_by_geometry(dataset: xr.Dataset,
         var = dataset[var_name]
         masked_vars[var_name] = var.where(mask)
 
-    masked_dataset = dataset.copy(data=masked_vars)
+    masked_dataset = xr.Dataset(masked_vars, coords=dataset.coords, attrs=dataset.attrs)
 
     _save_geometry_mask(masked_dataset, mask, save_geometry_mask)
     _save_geometry_wkt(masked_dataset, intersection_geometry, save_geometry_wkt)


Test patch:
diff --git a/test/util/test_geom.py b/test/util/test_geom.py
index e9707c54..d5da991c 100644
--- a/test/util/test_geom.py
+++ b/test/util/test_geom.py
@@ -5,6 +5,7 @@ import shapely.geometry
 import xarray as xr
 
 from xcube.api import new_cube
+from xcube.util.chunk import chunk_dataset
 from xcube.util.geom import get_dataset_geometry, get_dataset_bounds, get_geometry_mask, convert_geometry, \
     mask_dataset_by_geometry, clip_dataset_by_geometry
 
@@ -53,6 +54,20 @@ class DatasetGeometryTest(unittest.TestCase):
         self._assert_clipped_dataset_has_basic_props(cube)
         self._assert_dataset_mask_is_fine(cube, 'geom_mask')
 
+    def test_clip_dataset_for_chunked_input(self):
+        cube = chunk_dataset(self.cube, chunk_sizes=dict(time=1, lat=90, lon=90))
+        cube = clip_dataset_by_geometry(cube, self.triangle)
+        self._assert_clipped_dataset_has_basic_props(cube)
+        self.assertEqual(((1, 1, 1, 1, 1), (4,), (7,)), cube.temp.chunks)
+        self.assertEqual(((1, 1, 1, 1, 1), (4,), (7,)), cube.precip.chunks)
+
+    def test_mask_dataset_for_chunked_input(self):
+        cube = chunk_dataset(self.cube, chunk_sizes=dict(time=1, lat=90, lon=90))
+        cube = mask_dataset_by_geometry(cube, self.triangle)
+        self._assert_clipped_dataset_has_basic_props(cube)
+        self.assertEqual(((1, 1, 1, 1, 1), (4,), (7,)), cube.temp.chunks)
+        self.assertEqual(((1, 1, 1, 1, 1), (4,), (7,)), cube.precip.chunks)
+
     def _assert_clipped_dataset_has_basic_props(self, dataset):
         self.assertEqual({'time': 5, 'lat': 4, 'lon': 7}, dataset.dims)
         self.assertIn('temp', dataset)
Repository: taskforcesh/bullmq
Pull request #41 (2019-10-20 09:58:43)

Feat/support async backoffs


Patch:
diff --git a/src/classes/job.ts b/src/classes/job.ts
index fb9d8a8f26..5a0ab3533c 100644
--- a/src/classes/job.ts
+++ b/src/classes/job.ts
@@ -254,7 +254,7 @@ export class Job {
       const opts = queue.opts as WorkerOptions;
 
       // Check if backoff is needed
-      const delay = Backoffs.calculate(
+      const delay = await Backoffs.calculate(
         <BackoffOptions>this.opts.backoff,
         this.attemptsMade,
         opts.settings && opts.settings.backoffStrategies,


Test patch:
diff --git a/src/test/test_worker.ts b/src/test/test_worker.ts
index 10886d181d..30ffd557ca 100644
--- a/src/test/test_worker.ts
+++ b/src/test/test_worker.ts
@@ -1346,7 +1346,53 @@ describe('workers', function() {
       await queueScheduler.close();
     });
 
-    it.skip('should not retry a job that has been removed', async () => {
+    it('should be able to handle a custom backoff if it returns a promise', async function() {
+      this.timeout(12000);
+
+      const queueScheduler = new QueueScheduler(queueName);
+      await queueScheduler.waitUntilReady();
+
+      const worker = new Worker(
+        queueName,
+        async (job: Job) => {
+          if (job.attemptsMade < 2) {
+            throw new Error('some error');
+          }
+        },
+        {
+          settings: {
+            backoffStrategies: {
+              async custom() {
+                return delay(500);
+              },
+            },
+          },
+        },
+      );
+      const start = Date.now();
+      await queue.add(
+        'test',
+        { foo: 'bar' },
+        {
+          attempts: 3,
+          backoff: {
+            type: 'custom',
+          },
+        },
+      );
+
+      await new Promise(resolve => {
+        worker.on('completed', () => {
+          const elapse = Date.now() - start;
+          expect(elapse).to.be.greaterThan(1000);
+          resolve();
+        });
+      });
+
+      await worker.close();
+    });
+
+    it('should not retry a job that has been removed', async () => {
       const queueScheduler = new QueueScheduler(queueName);
       await queueScheduler.waitUntilReady();
 
@@ -1379,7 +1425,8 @@ describe('workers', function() {
             try {
               await job.retry();
             } catch (err) {
-              expect(err.message).to.equal(RetryErrors.JobNotExist);
+              // expect(err.message).to.equal(RetryErrors.JobNotExist);
+              expect(err.message).to.equal('Retried job not exist');
             }
 
             const completedCount = await queue.getCompletedCount();
@@ -1389,6 +1436,7 @@ describe('workers', function() {
           } catch (err) {
             reject(err);
           }
+          resolve();
         });
 
         worker.on('failed', failedHandler);
@@ -1396,7 +1444,7 @@ describe('workers', function() {
 
       await worker.close();
       await queueScheduler.close();
-    });
+    }).timeout(5000);
 
     it.skip('should not retry a job that has been retried already', async () => {
       let attempts = 0;
Repository: taskforcesh/bullmq
Pull request #74 (2019-11-20 20:32:18)

fix: avoid changing Job property types through JSON serialization/deserialization #70
renames Job#toJSON to Job#asJSON and adds a test


Patch:
diff --git a/src/classes/job.ts b/src/classes/job.ts
index 18d91202a8..ff3fab8fa1 100644
--- a/src/classes/job.ts
+++ b/src/classes/job.ts
@@ -144,7 +144,12 @@ export class Job {
     }
   }
 
-  toJSON(): JobJson {
+  toJSON() {
+    const { queue, ...withoutQueue } = this;
+    return withoutQueue;
+  }
+
+  asJSON(): JobJson {
     return {
       id: this.id,
       name: this.name,
@@ -452,7 +457,7 @@ export class Job {
   private addJob(client: IORedis.Redis): string {
     const queue = this.queue;
 
-    const jobData = this.toJSON();
+    const jobData = this.asJSON();
 
     return Scripts.addJob(client, queue, jobData, this.opts, this.id);
   }
diff --git a/src/classes/sandbox.ts b/src/classes/sandbox.ts
index 5d6f26ab48..aefb97a2c3 100644
--- a/src/classes/sandbox.ts
+++ b/src/classes/sandbox.ts
@@ -6,7 +6,7 @@ const sandbox = (processFile: any, childPool: any) => {
 
     child.send({
       cmd: 'start',
-      job: job.toJSON(),
+      job: job.asJSON(),
     });
 
     const done = new Promise((resolve, reject) => {


Test patch:
diff --git a/src/test/test_job.ts b/src/test/test_job.ts
index f81368efad..081b9b4f7c 100644
--- a/src/test/test_job.ts
+++ b/src/test/test_job.ts
@@ -67,6 +67,29 @@ describe('Job', function() {
     });
   });
 
+  describe('JSON.stringify', () => {
+    it('retains property types', async () => {
+      const data = { foo: 'bar' };
+      const job = await Job.create(queue, 'test', data);
+      job.returnvalue = 1;
+      job.progress = 20;
+      const json = JSON.stringify(job);
+      const parsed = JSON.parse(json);
+      expect(parsed).to.have.deep.property('data', data);
+      expect(parsed).to.have.property('name', 'test');
+      expect(parsed).to.have.property('returnvalue', 1);
+      expect(parsed).to.have.property('progress', 20);
+    });
+
+    it('omits the queue property to avoid a circular json error on node 8', async () => {
+      const data = { foo: 'bar' };
+      const job = await Job.create(queue, 'test', data);
+      const json = JSON.stringify(job);
+      const parsed = JSON.parse(json);
+      expect(parsed).not.to.have.property('queue');
+    });
+  });
+
   describe('.update', function() {
     it('should allow updating job data', async function() {
       const job = await Job.create(queue, 'test', { foo: 'bar' });
Repository: taskforcesh/bullmq
Pull request #177 (2020-04-10 15:33:42)

Fix/do not use global child pool


Patch:
diff --git a/src/classes/child-pool.ts b/src/classes/child-pool.ts
index 880d9fac37..0296d75b1f 100644
--- a/src/classes/child-pool.ts
+++ b/src/classes/child-pool.ts
@@ -114,5 +114,3 @@ export class ChildPool {
     return flatten(values(this.free));
   }
 }
-
-export const pool = new ChildPool();
diff --git a/src/classes/worker.ts b/src/classes/worker.ts
index 3c855269d1..50bdd56dd9 100644
--- a/src/classes/worker.ts
+++ b/src/classes/worker.ts
@@ -3,7 +3,7 @@ import { Redis } from 'ioredis';
 import * as path from 'path';
 import { Processor, WorkerOptions } from '../interfaces';
 import { QueueBase, Repeat } from './';
-import { ChildPool, pool } from './child-pool';
+import { ChildPool } from './child-pool';
 import { Job } from './job';
 import { RedisConnection } from './redis-connection';
 import sandbox from './sandbox';
@@ -72,7 +72,7 @@ export class Worker<T = any> extends QueueBase {
         throw new Error(`File ${processorFile} does not exist`);
       }
 
-      this.childPool = this.childPool || pool;
+      this.childPool = this.childPool || new ChildPool();
       this.processFn = sandbox(processor, this.childPool).bind(this);
     }
     this.timerManager = new TimerManager();


Test patch:
diff --git a/src/test/test_sandboxed_process.ts b/src/test/test_sandboxed_process.ts
index 59b3f1b594..48ca31c32f 100644
--- a/src/test/test_sandboxed_process.ts
+++ b/src/test/test_sandboxed_process.ts
@@ -1,7 +1,7 @@
 import { expect } from 'chai';
 import * as IORedis from 'ioredis';
 import { after } from 'lodash';
-import { Queue, QueueEvents, Worker, pool } from '@src/classes';
+import { Queue, QueueEvents, Worker } from '@src/classes';
 import { beforeEach } from 'mocha';
 import { v4 } from 'uuid';
 import { delay, removeAllQueueData } from '@src/utils';
@@ -22,7 +22,6 @@ describe('sandboxed process', () => {
   afterEach(async () => {
     await queue.close();
     await queueEvents.close();
-    pool.clean();
     await removeAllQueueData(new IORedis(), queueName);
   });
Repository: taskforcesh/bullmq
Pull request #205 (2020-05-24 22:03:06)

Fix/fix connection tests


Patch:
diff --git a/src/classes/redis-connection.ts b/src/classes/redis-connection.ts
index 66e5fa9e3b..f37753bdee 100644
--- a/src/classes/redis-connection.ts
+++ b/src/classes/redis-connection.ts
@@ -30,7 +30,7 @@ export class RedisConnection extends EventEmitter {
     this.initializing = this.init();
 
     this.initializing
-      .then(client => client.on('error', this.emit.bind(this, 'error')))
+      .then(client => client.on('error', err => this.emit('error', err)))
       .catch(err => this.emit('error', err));
   }
 


Test patch:
diff --git a/src/test/test_connection.ts b/src/test/test_connection.ts
index f9bb4320e3..e20298e10b 100644
--- a/src/test/test_connection.ts
+++ b/src/test/test_connection.ts
@@ -21,9 +21,6 @@ describe('connection', () => {
 
   it('should recover from a connection loss', async () => {
     let processor;
-    queue.on('error', () => {
-      // error event has to be observed or the exception will bubble up
-    });
 
     const processing = new Promise(resolve => {
       processor = async (job: Job) => {
@@ -34,6 +31,14 @@ describe('connection', () => {
 
     const worker = new Worker(queueName, processor);
 
+    worker.on('error', err => {
+      // error event has to be observed or the exception will bubble up
+    });
+
+    queue.on('error', err => {
+      // error event has to be observed or the exception will bubble up
+    });
+
     const workerClient = await worker.client;
     const queueClient = await queue.client;
 
@@ -71,6 +76,15 @@ describe('connection', () => {
     });
 
     const worker = new Worker(queueName, processor);
+
+    worker.on('error', err => {
+      // error event has to be observed or the exception will bubble up
+    });
+
+    queue.on('error', err => {
+      // error event has to be observed or the exception will bubble up
+    });
+
     await worker.waitUntilReady();
 
     worker.on('completed', async () => {
Repository: taskforcesh/bullmq
Pull request #572 (2021-05-31 10:38:46)

fix(obliterate): remove job logs


Patch:
diff --git a/src/commands/obliterate-2.lua b/src/commands/obliterate-2.lua
index 2a284ec3e4..5eb81a2123 100644
--- a/src/commands/obliterate-2.lua
+++ b/src/commands/obliterate-2.lua
@@ -35,6 +35,7 @@ end
 local function removeJobs(parentKey, keys)
     for i, key in ipairs(keys) do
         rcall("DEL", baseKey .. key)
+        rcall("DEL", baseKey .. key .. ':logs')
     end
     maxCount = maxCount - #keys
 end


Test patch:
diff --git a/src/test/test_obliterate.ts b/src/test/test_obliterate.ts
index 2b3efa839d..b577b83c9f 100644
--- a/src/test/test_obliterate.ts
+++ b/src/test/test_obliterate.ts
@@ -147,6 +147,24 @@ describe('Obliterate', () => {
     expect(keys.length).to.be.eql(0);
   });
 
+  it('should remove job logs', async () => {
+    const job = await queue.add('test', {});
+
+    const queueEvents = new QueueEvents(queue.name);
+
+    const worker = new Worker(queue.name, async job => {
+      return job.log('Lorem Ipsum Dolor Sit Amet');
+    });
+    await worker.waitUntilReady();
+
+    await job.waitUntilFinished(queueEvents);
+
+    await queue.obliterate({ force: true });
+
+    const { logs } = await queue.getJobLogs(job.id);
+    expect(logs).to.have.length(0);
+  });
+
   it('should obliterate a queue with high number of jobs in different statuses', async () => {
     const arr1 = [];
     for (let i = 0; i < 300; i++) {
Repository: taskforcesh/bullmq
Pull request #576 (2021-06-02 02:18:47)

Fix getDependencies to parse processed values
I'm curious if this was intended or not, since the values are not parsed and from getChildrenValues are parsed, so this pr tries also to parse processed values from getDependencies method


Patch:
diff --git a/src/classes/job.ts b/src/classes/job.ts
index 185dfadd81..a133782511 100644
--- a/src/classes/job.ts
+++ b/src/classes/job.ts
@@ -560,7 +560,14 @@ export class Job<T = any, R = any, N extends string = string> {
       [null | Error, string[]],
     ];
 
-    return { processed, unprocessed };
+    const transformedProcessed = Object.entries(processed).reduce(
+      (accumulator, [key, value]) => {
+        return { ...accumulator, [key]: JSON.parse(value) };
+      },
+      {},
+    );
+
+    return { processed: transformedProcessed, unprocessed };
   }
 
   /**


Test patch:
diff --git a/src/test/test_flow.ts b/src/test/test_flow.ts
index 41f722b950..59a0dbd7dc 100644
--- a/src/test/test_flow.ts
+++ b/src/test/test_flow.ts
@@ -289,6 +289,7 @@ describe('flows', () => {
 
           const jobKey = queue.toKey(tree.children[0].job.id);
           expect(childrenValues[jobKey]).to.be.deep.equal(values[0]);
+          expect(processed[jobKey]).to.be.deep.equal(values[0]);
 
           resolve();
         } catch (err) {
diff --git a/src/test/test_worker.ts b/src/test/test_worker.ts
index 56fb914e88..ab8186081a 100644
--- a/src/test/test_worker.ts
+++ b/src/test/test_worker.ts
@@ -1833,7 +1833,7 @@ describe('workers', function() {
       );
 
       expect(processed2).to.deep.equal({
-        [`bull:${queueName}:${child1.id}`]: `"return value1"`,
+        [`bull:${queueName}:${child1.id}`]: 'return value1',
       });
       expect(unprocessed2).to.have.length(2);
       expect(movedToWaitingChildren).to.be.true;
@@ -1850,8 +1850,8 @@ describe('workers', function() {
       const isWaitingChildren1 = await parent.isWaitingChildren();
 
       expect(processed3).to.deep.equal({
-        [`bull:${queueName}:${child1.id}`]: `"return value1"`,
-        [`bull:${queueName}:${child2.id}`]: `"return value2"`,
+        [`bull:${queueName}:${child1.id}`]: 'return value1',
+        [`bull:${queueName}:${child2.id}`]: 'return value2',
       });
       expect(unprocessed3).to.have.length(1);
       expect(isWaitingChildren1).to.be.true;
@@ -1871,9 +1871,9 @@ describe('workers', function() {
       );
 
       expect(processed4).to.deep.equal({
-        [`bull:${queueName}:${child1.id}`]: `"return value1"`,
-        [`bull:${queueName}:${child2.id}`]: `"return value2"`,
-        [`bull:${queueName}:${child3.id}`]: `"return value3"`,
+        [`bull:${queueName}:${child1.id}`]: 'return value1',
+        [`bull:${queueName}:${child2.id}`]: 'return value2',
+        [`bull:${queueName}:${child3.id}`]: 'return value3',
       });
       expect(unprocessed4).to.have.length(0);
       expect(isWaitingChildren2).to.be.false;
Repository: taskforcesh/bullmq
Pull request #586 (2021-06-08 23:41:53)

Fix typo queue options


Patch:
diff --git a/src/interfaces/queue-options.ts b/src/interfaces/queue-options.ts
index e6a21c3212..c52ea475c1 100644
--- a/src/interfaces/queue-options.ts
+++ b/src/interfaces/queue-options.ts
@@ -24,7 +24,7 @@ export interface QueueOptions extends QueueBaseOptions {
   streams?: {
     events: {
       /**
-       * Max aproximated length for streams
+       * Max approximated length for streams
        */
       maxLen: number;
     };


Test patch:
diff --git a/src/test/test_worker.ts b/src/test/test_worker.ts
index d64a5887b3..04cf328dff 100644
--- a/src/test/test_worker.ts
+++ b/src/test/test_worker.ts
@@ -8,7 +8,7 @@ import { after, times, once } from 'lodash';
 import { RetryErrors } from '../enums';
 import * as sinon from 'sinon';
 
-describe.only('workers', function() {
+describe('workers', function() {
   const sandbox = sinon.createSandbox();
 
   let queue: Queue;
Repository: taskforcesh/bullmq
Pull request #592 (2021-06-11 04:53:53)

Feat expose queueName in job
As we have getFlow method, we can get a tree for an specific parent node, but in order to know from which queue the children and grand children belong, it would be nice to expose the name of the queue from the job instance.


Patch:
diff --git a/docs/gitbook/guide/retrying-failing-jobs.md b/docs/gitbook/guide/retrying-failing-jobs.md
index 09e88d65e6..def98e8755 100644
--- a/docs/gitbook/guide/retrying-failing-jobs.md
+++ b/docs/gitbook/guide/retrying-failing-jobs.md
@@ -28,6 +28,8 @@ await queue.add(
 
 The current built-in backoff functions are "exponential" and "fixed".
 
+With an exponential backoff, it will retry after `2 ^ attempts * delay` milliseconds. For example, with a delay of 3000 milliseconds, for the 7th attempt, it will retry 2^7 * 3000 milliseconds = 6.4 minutes after the previous attempt. With a fixed backoff, it will retry after `delay` milliseconds, so with a delay of 3000 milliseconds, it will retry _every_ attempt 3000 milliseconds after the previous attempt.
+
 If you want to define your custom backoff you need to define it at the worker:
 
 ```typescript
diff --git a/src/classes/job.ts b/src/classes/job.ts
index b31b1dce07..6ee105e79e 100644
--- a/src/classes/job.ts
+++ b/src/classes/job.ts
@@ -524,6 +524,10 @@ export class Job<T = any, R = any, N extends string = string> {
     return (await this.isInList('wait')) || (await this.isInList('paused'));
   }
 
+  get queueName(): string {
+    return this.queue.name;
+  }
+
   /**
    * @method getState
    * Get current state.
diff --git a/src/classes/scripts.ts b/src/classes/scripts.ts
index b98a6f035d..04d105f5ec 100644
--- a/src/classes/scripts.ts
+++ b/src/classes/scripts.ts
@@ -22,6 +22,7 @@ import { RedisClient } from './redis-connection';
 
 export type MinimalQueue = Pick<
   QueueBase,
+  | 'name'
   | 'client'
   | 'toKey'
   | 'keys'


Test patch:
diff --git a/src/test/test_flow.ts b/src/test/test_flow.ts
index f1384d1a74..b7d3191f39 100644
--- a/src/test/test_flow.ts
+++ b/src/test/test_flow.ts
@@ -243,9 +243,11 @@ describe('flows', () => {
 
     expect(children[0].job.id).to.be.ok;
     expect(children[0].job.data.foo).to.be.eql('bar');
+    expect(children[0].job.queueName).to.be.eql(queueName);
     expect(children[0].children).to.have.length(1);
 
     expect(children[0].children[0].job.id).to.be.ok;
+    expect(children[0].children[0].job.queueName).to.be.eql(queueName);
     expect(children[0].children[0].job.data.foo).to.be.eql('baz');
 
     expect(children[0].children[0].children[0].job.id).to.be.ok;
Repository: taskforcesh/bullmq
Pull request #857 (2021-11-04 00:51:52)

perf(added): remove data and opts from event
BREAKING CHANGE: remove data and opts from added event

manast: Something is not right with this PR, there are a lot of changes from autogenerated doc files.
roggervalf: > Something is not right with this PR, there are a lot of changes from autogenerated doc files.

yeah, that was because next is not updated with the last master version

Patch:
diff --git a/src/classes/queue-events.ts b/src/classes/queue-events.ts
index b8284530f4..658eb25536 100644
--- a/src/classes/queue-events.ts
+++ b/src/classes/queue-events.ts
@@ -23,7 +23,7 @@ export interface QueueEventsListener {
    * This event is triggered when a job is created.
    */
   added: (
-    args: { jobId: string; name: string; data: string; opts: string },
+    args: { jobId: string; name: string; },
     id: string,
   ) => void;
 
diff --git a/src/commands/addJob-9.lua b/src/commands/addJob-9.lua
index ab17034e43..748ccf747c 100644
--- a/src/commands/addJob-9.lua
+++ b/src/commands/addJob-9.lua
@@ -103,8 +103,7 @@ else
     "timestamp", timestamp, "delay", delay, "priority", priority )
 end
 
--- TODO: do not send data and opts to the event added (for performance reasons).
-rcall("XADD", KEYS[8], "*", "event", "added", "jobId", jobId, "name", args[3], "data", ARGV[2], "opts", jsonOpts)
+rcall("XADD", KEYS[8], "*", "event", "added", "jobId", jobId, "name", args[3])
 
 -- Check if job is delayed
 local delayedTimestamp = (delay > 0 and (timestamp + delay)) or 0


Test patch:
diff --git a/tests/test_events.ts b/tests/test_events.ts
index 2c60a3a54a..838e744e73 100644
--- a/tests/test_events.ts
+++ b/tests/test_events.ts
@@ -202,14 +202,11 @@ describe('events', function () {
       connection,
     });
     const testName = 'test';
-    const testData = { foo: 'bar' };
 
     const added = new Promise<void>(resolve => {
-      queueEvents.once('added', ({ jobId, name, data, opts }) => {
+      queueEvents.once('added', ({ jobId, name }) => {
         expect(jobId).to.be.equal('1');
         expect(name).to.be.equal(testName);
-        expect(data).to.be.equal(JSON.stringify(testData));
-        expect(JSON.parse(opts)).to.be.deep.equal({ attempts: 0, delay: 0 });
         resolve();
       });
     });