manager: Add unique workers batches and task report handling

apps/task-poster/src/fetcher.ts

@@ -80,6 +80,10 @@ export type Fetcher = {
 
   // posting schedule: whitelist of days/time windows when tasks can be posted
   schedule?: Schedule;
+
+  // repetitions: how many tasks a single worker may complete from each import
+  // 0 means no limit, 1 means each worker may do one task.
+  repetitions?: number;
 };
 
 const api = axios.create({
@@ -186,8 +190,10 @@ export const fetcherForm = async (
 
       <label for="type"><strong>Data source</strong><br/>
       <small>How new tasks will be fetched.</small></label>
-      <select name="type" id="type">${["csv", "pipeline", "constant"].map(a =>
-	`<option value="${a}" ${values.type == a ? "selected" : ""}>${a}</option>`).join("")}
+      <select name="type" id="type"
+        onchange="repetitions.disabled=this.value!=='csv';if(repetitions.disabled)repetitions.value=0">
+        ${["csv", "pipeline", "constant"].map(a =>
+	        `<option value="${a}" ${values.type == a ? "selected" : ""}>${a}</option>`).join("")}
       </select>
 
   </fieldset>
@@ -239,6 +245,15 @@ export const fetcherForm = async (
 	</div>
       </section>
 
+      <div class="mt">
+        <label for="repetitions"><strong>Repetitions per worker</strong><br/>
+        <small>How many tasks a single worker may complete per import batch.
+        0 = no limit, 1 = each worker may do one task. Only applies to CSV data sources.</small></label>
+        <input type="number" id="repetitions" name="repetitions" min="0" step="1"
+          value="${values.repetitions ?? 0}"
+          ${(values.type ?? 'csv') !== 'csv' ? "disabled" : ""} />
+      </div>
+
       <div class="mt"></div>
       <label for="capabilities"><strong>Capabilities</strong><br/>
       <small>Comma separated list of capabilities required for tasks (note: only 1 capability is used at the moment).</small></label>
@@ -490,6 +505,7 @@ export const createFetcher = async (
 
     status: oldFetcher?.status ?? "active",
     hidden: fields.hidden === "on" || fields.hidden === true,
+    repetitions: Number(fields.repetitions) || 0,
 
     schedule: (() => {
       try {
@@ -586,7 +602,7 @@ export const getTasks = async (fetcher: Fetcher, csv: string) => {
 	}
 
 	// regex filter
-	if (!task.result || (regex && regex.test(task.result)) || task.result == "<TASK REPORTED AND SKIPPED>") {
+	if (!task.result || (regex && regex.test(task.result)) || task.type === "report" || task.result == "<TASK REPORTED AND SKIPPED>") {
 	  continue;
 	}
 
@@ -614,6 +630,8 @@ export const getTasks = async (fetcher: Fetcher, csv: string) => {
   // avoid expensive mistakes.
   const maxPrice = BigInt(fetcher.price) * BigInt(10);
 
+  const batchRunId = ulid();
+
   const tasks = data.map(
     (d, _idx) => {
       const rawPrice = d["dataffect/reward"] ?
@@ -628,6 +646,8 @@ export const getTasks = async (fetcher: Fetcher, csv: string) => {
 	templateId: fetcher.template,
 	templateData: JSON.stringify(d),
 	capability: fetcher.capabilities[0],
+	batchId: `${fetcher.datasetId}-${fetcher.index}-${batchRunId}`,
+	repetitions: fetcher.repetitions ?? 0,
       } as Task;
     },
   );
@@ -886,7 +906,7 @@ export const processResults = async (f: Fetcher, batchSize: number) => {
 
     let importCount = 0;
     for (const d of data as any) {
-      if (d.type !== "submission")
+      if (d.type !== "submission" && d.type !== "report")
 	continue;
       db.beginTransaction();
       await db.delete([...keyBase, "active", d.taskId]);
@@ -972,6 +992,15 @@ const formValidations: ValidationMap = {
   },
 
   template: (v: string) => (!v || v.length === 0) && "Template is required",
+  repetitions: (v: any) => {
+    if (v === undefined || v === "") return false;
+    const repetitions = Number(v);
+    return (
+      (isNaN(repetitions) && "Repetitions must be a number") ||
+      (!Number.isInteger(repetitions) && "Repetitions must be a whole number") ||
+      (repetitions < 0 && "Repetitions cannot be negative")
+    );
+  },
   timeLimitSeconds: (v: any) => {
     const num = Number(v);
     return (
@@ -1075,10 +1104,11 @@ export const addFetcherRoutes = (app: Express): void => {
   <li>Finished: ${doneSize}</li>
   <li>Failed: ${failedSize}</li>
   <li>Batch / Freq: ${f.batchSize} / ${f.frequency}</li>
+  ${f.repetitions ? `<li>Repetitions per worker: ${f.repetitions}</li>` : ''}
   <li>Time Limit: ${f.timeLimitSeconds}s</li>
   <li>Schedule: ${f.schedule && Object.keys(f.schedule).length > 0
     ? Object.entries(f.schedule).map(([day, slots]) =>
-        `${day}: ${(slots as TimeSlot[]).map(s => s.from && s.to ? `${s.from}-${s.to}` : 'all day').join(', ')}`
+        `${day}: ${Array.isArray(slots) ? (slots as TimeSlot[]).map(s => s.from && s.to ? `${s.from}-${s.to}` : 'all day').join(', ') : 'all day'}`
       ).join('; ')
     : 'continuous (no restrictions)'}
 

apps/worker-app/app/composables/useTasks.ts

@@ -37,47 +37,28 @@ export const useTasks = () => {
 
   const useGetActiveTasks = (index: Ref<number | string>) => {
     const queryClient = useQueryClient();
-
-    const handleTaskEvent = (event: CustomEvent<Task>) => {
-      queryClient.setQueryData<Task[]>(["tasks", index], (old) =>
-        old ? [...old, event.detail] : [event.detail],
-      );
-    };
+    const { account } = useAuth();
 
     const invalidateTasks = () => {
-      queryClient.invalidateQueries({ queryKey: ["tasks", index] });
+      queryClient.invalidateQueries({ queryKey: ["tasks", index, account] });
     };
 
-    onMounted(() => {
-      instance.value?.events.addEventListener("task:created", handleTaskEvent);
-      instance.value?.events.addEventListener("task:rejected", invalidateTasks);
-      instance.value?.events.addEventListener(
-        "task:completed",
-        invalidateTasks,
-      );
-      instance.value?.events.addEventListener(
-        "task:expired",
-        invalidateTasks,
-      );
-    });
-
-    onUnmounted(() => {
-      instance.value?.events.removeEventListener(
-        "task:created",
-        handleTaskEvent,
-      );
-      instance.value?.events.removeEventListener(
-        "task:rejected",
-        invalidateTasks,
-      );
-      instance.value?.events.removeEventListener(
-        "task:completed",
-        invalidateTasks,
-      );
-      instance.value?.events.removeEventListener(
-        "task:expired",
-        invalidateTasks,
-      );
+    // Register listeners as soon as instance is available, not just on mount.
+    watchEffect((onCleanup) => {
+      const events = instance.value?.events;
+      if (!events) return;
+
+      events.addEventListener("task:created", invalidateTasks);
+      events.addEventListener("task:rejected", invalidateTasks);
+      events.addEventListener("task:completed", invalidateTasks);
+      events.addEventListener("task:expired", invalidateTasks);
+
+      onCleanup(() => {
+        events.removeEventListener("task:created", invalidateTasks);
+        events.removeEventListener("task:rejected", invalidateTasks);
+        events.removeEventListener("task:completed", invalidateTasks);
+        events.removeEventListener("task:expired", invalidateTasks);
+      });
     });
 
     return useGetTasks(index);

core/protobufs/src/effect.ts

@@ -2766,6 +2766,8 @@ export interface Task {
   templateId: string
   templateData: string
   capability?: string
+  batchId?: string
+  repetitions: number
 }
 
 export namespace Task {
@@ -2813,6 +2815,16 @@ export namespace Task {
           w.string(obj.capability)
         }
 
+        if (obj.batchId != null) {
+          w.uint32(66)
+          w.string(obj.batchId)
+        }
+
+        if ((obj.repetitions != null && obj.repetitions !== 0)) {
+          w.uint32(72)
+          w.uint32(obj.repetitions)
+        }
+
         if (opts.lengthDelimited !== false) {
           w.ldelim()
         }
@@ -2823,7 +2835,8 @@ export namespace Task {
           reward: 0n,
           timeLimitSeconds: 0,
           templateId: '',
-          templateData: ''
+          templateData: '',
+          repetitions: 0
         }
 
         const end = length == null ? reader.len : reader.pos + length
@@ -2860,6 +2873,14 @@ export namespace Task {
               obj.capability = reader.string()
               break
             }
+            case 8: {
+              obj.batchId = reader.string()
+              break
+            }
+            case 9: {
+              obj.repetitions = reader.uint32()
+              break
+            }
             default: {
               reader.skipType(tag & 7)
               break

core/protobufs/src/task/task.proto

@@ -16,6 +16,8 @@ message Task {
   string template_id = 5;
   string template_data = 6;
   optional string capability = 7;
+  optional string batch_id = 8;
+  uint32 repetitions = 9;
 }
 
 message TaskAssignment {

core/protocol/src/consts.ts

@@ -1,4 +1,4 @@
-export const TASK_ACCEPTANCE_TIME = 1000 * 60 * 10; // 10 minutes
-
+// 10 minutes, in milliseconds. Same duration as modules/manager/src/consts.ts
+export const TASK_ACCEPTANCE_TIME = 1000 * 60 * 10;
 export const PROTOCOL_NAME = "effectai";
 export const PROTOCOL_VERSION = "0.0.2";

modules/manager/src/consts.ts

@@ -1,4 +1,5 @@
-export const TASK_ACCEPTANCE_TIME = 600;
+// 10 minutes, in seconds. Same duration as core/protocol/src/consts.ts
+export const TASK_ACCEPTANCE_TIME = 60 * 10;
 export const ACTIVE_TASK_TRESHOLD = 50;
 export const PAYMENT_BATCH_SIZE = 60;
 export const PAYMENT_VERSION = 1;

modules/manager/src/main.ts

@@ -387,12 +387,24 @@ export const createManager = async ({
           taskId,
           index: "completed",
         })
-        .then(
-          (a) =>
-            a.events
-              .filter((e: any) => e.type === "submission")
-              .map((e: any) => ({ ...e, taskId }))[0],
-        )
+        .then((taskRecord) => {
+          const event = taskRecord.events.find(
+            (taskEvent: any) =>
+              taskEvent.type === "submission" || taskEvent.type === "report",
+          );
+          if (!event) return { taskId, error: "NOT FOUND" };
+
+          if (event.type === "report") {
+            return {
+              type: "report",
+              timestamp: event.timestamp,
+              taskId,
+              submissionByPeer: event.reportedByPeer,
+              result: event.result,
+            };
+          }
+          return { ...event, taskId };
+        })
         .catch((_e) => ({ taskId, error: "NOT FOUND" }));
     });