Allow specifying the number of tasks to retrieve for a client (#385)

* Allow specifying the number of tasks to retrieve for a client in ClientGetTask method

* Maintain compatibility for older clients by limiting task retrieval to one for versions 1.0 and below

* Update Unregister method to unassign tasks before client removal; add SQL schema change for Publisher column

* Add Verifying status to task enumeration; update SQL schema for Signatures_Roms

* Update Unregister method to unassign tasks based on multiple current statuses

Author: michael-j-green

hasheous-lib/Classes/TaskManagement.cs

@@ -258,21 +258,42 @@ public async static Task UpdateClient(string clientAPIKey, string publicId, stri
         }
 
         /// <summary>
-        /// Retrieves the next available task for the specified client, or the currently assigned one if it exists.
+        /// Retrieves the next available tasks for the specified client. Returns unassigned pending tasks and any tasks already assigned to this client in Assigned or InProgress status.
         /// </summary>
         /// <param name="clientAPIKey">The API key of the client.</param>
         /// <param name="publicId">The public client ID.</param>
-        public async static Task<QueueItemModel?> ClientGetTask(string clientAPIKey, string publicId)
+        /// <param name="numberOfTasks">The number of tasks to retrieve (optional).</param>
+        public async static Task<object?> ClientGetTask(string clientAPIKey, string publicId, int numberOfTasks = 1)
         {
+            if (numberOfTasks <= 0)
+            {
+                numberOfTasks = 1; // default to 1 task if invalid number provided
+            }
+            else if (numberOfTasks > 20)
+            {
+                numberOfTasks = 20; // cap at 20 tasks to prevent overload
+            }
+
             ClientModel? client = await GetClientByAPIKeyAndPublicId(clientAPIKey, publicId);
             if (client == null)
             {
                 throw new Exception("Invalid client API key or public ID.");
             }
 
+            // clients older than 1.1.0 expect a single task back
+            if (client.ClientVersion != null)
+            {
+                Version clientVersion = new Version(client.ClientVersion);
+                if (clientVersion.Major == 1 && clientVersion.Minor == 0)
+                {
+                    numberOfTasks = 1;
+                }
+            }
+
             // Use a transaction with SELECT FOR UPDATE SKIP LOCKED to prevent race conditions
             // when multiple clients request jobs simultaneously.
             // The FOR UPDATE SKIP LOCKED ensures that each concurrent client gets a different task.
+            // Selects: unassigned pending tasks (status=0, client_id IS NULL) OR tasks assigned to this client in Assigned/InProgress status (status=10/20, client_id=@client_id)
             DateTime now = DateTime.UtcNow;
 
             // Step 1: SELECT with FOR UPDATE SKIP LOCKED (locks the row)
@@ -281,8 +302,7 @@ public async static Task UpdateClient(string clientAPIKey, string publicId, stri
             string selectSql = @"SELECT tq.id AS id, tq.create_time AS create_time, tq.dataobjectid AS dataobjectid, tq.task_name AS task_name, tq.status AS status, tq.client_id AS client_id, tq.parameters AS parameters, tq.result AS result, tq.error_message AS error_message, tq.start_time AS start_time, tq.completion_time AS completion_time
                 FROM Task_Queue tq
                 LEFT JOIN Task_Queue_Capabilities tqc ON tq.id = tqc.task_queue_id
-                WHERE tq.status = 0
-                AND (tq.client_id IS NULL OR tq.client_id = @client_id)
+                WHERE ((tq.status = 0 AND tq.client_id IS NULL) OR ((tq.status = 10 OR tq.status = 20) AND tq.client_id = @client_id))
                 AND NOT EXISTS (
                     SELECT 1 
                     FROM Task_Queue_Capabilities tqc_required
@@ -291,7 +311,7 @@ AND tqc_required.capability_id NOT IN (" + string.Join(", ", client.Capabilities
                 )
                 GROUP BY tq.id
                 ORDER BY tq.create_time ASC
-                LIMIT 1
+                LIMIT " + numberOfTasks + @"
                 FOR UPDATE SKIP LOCKED;";
 
             // UPDATE using the same logic to identify the task
@@ -308,8 +328,7 @@ SELECT id FROM (
                         SELECT tq.id
                         FROM Task_Queue tq
                         LEFT JOIN Task_Queue_Capabilities tqc ON tq.id = tqc.task_queue_id
-                        WHERE tq.status = 0
-                        AND (tq.client_id IS NULL OR tq.client_id = @client_id)
+                        WHERE ((tq.status = 0 AND tq.client_id IS NULL) OR ((tq.status = 10 OR tq.status = 20) AND tq.client_id = @client_id))
                         AND NOT EXISTS (
                             SELECT 1 
                             FROM Task_Queue_Capabilities tqc_required
@@ -318,7 +337,7 @@ AND tqc_required.capability_id NOT IN (" + string.Join(", ", client.Capabilities
                         )
                         GROUP BY tq.id
                         ORDER BY tq.create_time ASC
-                        LIMIT 1
+                        LIMIT " + numberOfTasks + @"
                         FOR UPDATE SKIP LOCKED
                     ) AS subquery
                 );";
@@ -348,18 +367,34 @@ FOR UPDATE SKIP LOCKED
                 return null;
             }
 
-            QueueItemModel task = new QueueItemModel(dt.Rows[0]);
+            List<QueueItemModel> tasks = new List<QueueItemModel>();
+            foreach (DataRow row in dt.Rows)
+            {
+                QueueItemModel task = new QueueItemModel(row);
+
+                // The task has already been updated by the transaction
+                // Update the local object to reflect the database state
+                task.ClientId = client.Id;
+                task.Status = QueueItemStatus.Assigned;
+                task.StartedAt = now;
+                task.CompletedAt = now;
+                task.Result = "";
+                task.ErrorMessage = "";
+
+                // clients older than 1.1.0 expect a single task back
+                if (client.ClientVersion != null)
+                {
+                    Version clientVersion = new Version(client.ClientVersion);
+                    if (clientVersion.Major == 1 && clientVersion.Minor == 0)
+                    {
+                        return task; // only return the first task for older clients
+                    }
+                }
 
-            // The task has already been updated by the transaction
-            // Update the local object to reflect the database state
-            task.ClientId = client.Id;
-            task.Status = QueueItemStatus.Assigned;
-            task.StartedAt = now;
-            task.CompletedAt = now;
-            task.Result = "";
-            task.ErrorMessage = "";
+                tasks.Add(task);
+            }
 
-            return task;
+            return tasks;
         }
 
         /// <summary>

hasheous-lib/Models/TaskClientModel.cs

@@ -250,6 +250,18 @@ public ClientModel Refresh(DataRow? row = null)
         /// </summary>
         public async Task Unregister()
         {
+            // update the database to unassign any tasks currently assigned to this client
+            await Config.database.ExecuteCMDAsync("UPDATE Task_Queue SET client_id = NULL, status = @status WHERE client_id = @id AND status IN (@current_status1, @current_status2, @current_status3, @current_status4, @current_status5)", new Dictionary<string, object>
+            {
+                { "@id", this.Id },
+                { "@status", QueueItemStatus.Pending },
+                { "@current_status1", QueueItemStatus.Pending },
+                { "@current_status2", QueueItemStatus.Assigned },
+                { "@current_status3", QueueItemStatus.Verifying },
+                { "@current_status4", QueueItemStatus.Failed },
+                { "@current_status5", QueueItemStatus.Cancelled }
+            });
+
             // update the database to remove this client
             await Config.database.ExecuteCMDAsync("DELETE FROM Task_Clients WHERE id = @id", new Dictionary<string, object>
             { { "@id", this.Id } });

hasheous-lib/Models/TaskQueueModel.cs

@@ -268,7 +268,11 @@ public enum QueueItemStatus
         /// <summary>
         /// The task has been cancelled.
         /// </summary>
-        Cancelled = 60
+        Cancelled = 60,
+        /// <summary>
+        /// The task is currently being verified.
+        /// </summary>
+        Verifying = 70
     }
 
     /// <summary>

hasheous-lib/Schema/hasheous-1032.sql

@@ -0,0 +1,6 @@
+ALTER TABLE `Signatures_Publishers`
+CHANGE `Publisher` `Publisher` varchar(255) DEFAULT NULL;
+
+ALTER TABLE `Signatures_Roms`
+CHANGE `Languages` `Languages` LONGTEXT DEFAULT NULL,
+CHANGE `Countries` `Countries` LONGTEXT DEFAULT NULL;

hasheous/Controllers/V1.0/TaskWorkerController.cs

@@ -197,14 +197,15 @@ public async Task<IActionResult> ClientHeartbeat(string publicid)
         /// Retrieves the next job assigned to the specified task worker client. If a job is already assigned, it returns that job.
         /// </summary>
         /// <param name="publicid">The public identifier of the client requesting a job.</param>
+        /// <param name="numberOfTasks">The number of tasks to retrieve (optional, default is 1).</param>
         /// <returns>An IActionResult containing the job details or null if no job is available.</returns>
         [MapToApiVersion("1.0")]
         [HttpGet]
         [Authentication.TaskWorkerAPIKey.TaskWorkerAPIKey()]
         [ProducesResponseType(StatusCodes.Status200OK)]
         [ProducesResponseType(StatusCodes.Status404NotFound)]
         [Route("clients/{publicid}/job")]
-        public async Task<IActionResult> GetJobForClient(string publicid)
+        public async Task<IActionResult> GetJobForClient(string publicid, [FromQuery] int numberOfTasks = 1)
         {
             // get the api key from the header
             string? apiKey = Request.Headers.TryGetValue(Authentication.TaskWorkerAPIKey.APIKeyHeaderName, out var headerValue) ? headerValue.FirstOrDefault() : null;
@@ -213,7 +214,7 @@ public async Task<IActionResult> GetJobForClient(string publicid)
                 return Unauthorized("API key is missing.");
             }
 
-            var job = await ClientManagement.ClientGetTask(apiKey, publicid);
+            var job = await ClientManagement.ClientGetTask(apiKey, publicid, numberOfTasks);
 
             return Ok(job);
         }