fix: deduplicate list of clients in retrieval tasks (#513)

Recently, we discovered that there is a lot of deals using the same
payload CID `bafkqaaa`. There are more than 22k deals, but only
between 27 miners and 9 clients. As a result, when one of such deals is
picked for testing, the list of clients is very long and full of
duplicates.

In this change, I am modifying the query sampling eligible deals
to de-duplicate the list of clients in each task.

Signed-off-by: Miroslav Bajtoš <oss@bajtos.net>

Author: bajtos

api/lib/round-tracker.js

@@ -307,7 +307,7 @@ export async function maybeCreateSparkRound (pgClient, {
   }
 }
 
-async function defineTasksForRound (pgClient, sparkRoundNumber, taskCount) {
+export async function defineTasksForRound (pgClient, sparkRoundNumber, taskCount) {
   await pgClient.query(`
     INSERT INTO retrieval_tasks (round_id, cid, miner_id, clients)
     WITH selected AS (
@@ -317,7 +317,7 @@ async function defineTasksForRound (pgClient, sparkRoundNumber, taskCount) {
       ORDER BY random()
       LIMIT $2
     )
-    SELECT $1 as round_id, selected.payload_cid as cid, selected.miner_id, array_agg(client_id) as clients
+    SELECT $1 as round_id, selected.payload_cid as cid, selected.miner_id, array_agg(DISTINCT client_id) as clients
     FROM selected
     LEFT JOIN eligible_deals
     ON selected.payload_cid = eligible_deals.payload_cid AND selected.miner_id = eligible_deals.miner_id

api/test/round-tracker.test.js

@@ -5,6 +5,7 @@ import {
   BASELINE_TASKS_PER_NODE,
   TASKS_EXECUTED_PER_ROUND,
   ROUND_TASKS_TO_NODE_TASKS_RATIO,
+  defineTasksForRound,
   getRoundStartEpoch,
   getRoundStartEpochWithBackoff,
   mapCurrentMeridianRoundToSparkRound,
@@ -560,6 +561,72 @@ describe('Round Tracker', () => {
         assert.strictEqual(sparkRound.max_tasks_per_node, MAX_TASKS_PER_NODE_LIMIT)
       })
     })
+
+    describe('defineTasksForRound', () => {
+      before(async () => {
+        // Mark all existing deals as expired
+        await pgClient.query(`
+          UPDATE eligible_deals SET expires_at = NOW() - INTERVAL '1 day'
+        `)
+      })
+
+      after(async () => {
+        // Revert the change that expired existing deals
+        await pgClient.query(`
+          UPDATE eligible_deals SET expires_at = NOW() + INTERVAL '1 year'
+        `)
+      })
+
+      it('merges duplicate clients', async () => {
+        // Delete any eligible deals created by previous test runs
+        await pgClient.query(`
+          DELETE FROM eligible_deals WHERE client_id = 'f0050'
+        `)
+
+        // Create deals from the same client. First two deals are with the same SP, the third is not.
+        // All deals have the same payload_cid.
+        // Only these deals will be available for sampling
+        await pgClient.query(`
+          INSERT INTO eligible_deals
+          (miner_id, client_id, piece_cid, piece_size, payload_cid, expires_at, sourced_from_f05_state)
+          VALUES
+          ('f0010', 'f0050', 'baga1', 1, 'bafkqaaa', NOW() + INTERVAL '1 year', true),
+          ('f0010', 'f0050', 'baga2', 1, 'bafkqaaa', NOW() + INTERVAL '1 year', true),
+          ('f0011', 'f0050', 'baga1', 1, 'bafkqaaa', NOW() + INTERVAL '1 year', true)
+        `)
+
+        // Create a new round and define tasks for the round
+        const roundId = 1
+        await pgClient.query(`
+          INSERT INTO spark_rounds
+          (id, created_at, meridian_address, meridian_round, start_epoch, max_tasks_per_node)
+          VALUES
+          ($1, NOW(), '0x1a', 1, 1, 15)
+        `, [
+          roundId
+        ])
+        await defineTasksForRound(pgClient, roundId, 3)
+
+        const { rows: tasks } = await pgClient.query(
+          'SELECT miner_id, cid, clients FROM retrieval_tasks WHERE round_id = $1',
+          [roundId]
+        )
+
+        assert.deepStrictEqual(tasks, [
+          {
+            cid: 'bafkqaaa',
+            miner_id: 'f0010',
+            // Important: clients are deduplicated
+            clients: ['f0050']
+          },
+          {
+            cid: 'bafkqaaa',
+            miner_id: 'f0011',
+            clients: ['f0050']
+          }
+        ])
+      })
+    })
   })
 
   describe('getRoundStartEpoch', () => {