some more cleanup

Author: juliusgeo

…2_batching_consolidated_v1_0_111.sql.sql …0_batching_consolidated_v1_0_118.sql.sqlcmd/hatchet-migrate/migrate/migrations/20260531213452_batching_consolidated_v1_0_111.sql.sql renamed to cmd/hatchet-migrate/migrate/migrations/20260612000000_batching_consolidated_v1_0_118.sql.sql cmd/hatchet-migrate/migrate/migrations/20260531213452_batching_consolidated_v1_0_111.sql.sql renamed to cmd/hatchet-migrate/migrate/migrations/20260612000000_batching_consolidated_v1_0_118.sql.sql

@@ -509,8 +509,8 @@ func (d *DispatcherImpl) handleTaskBulkAssignedTask(ctx context.Context, msg *ms
 		// load the step runs from the database
 		taskIds := make([]int64, 0)
 
-		for _, ids := range innerMsg.WorkerIdToTaskIds {
-			taskIds = append(taskIds, ids...)
+		for _, tasks := range innerMsg.WorkerIdToTaskIds {
+			taskIds = append(taskIds, tasks...)
 		}
 
 		taskIdToData, err := d.populateTaskData(ctx, requeue, msg.TenantID, taskIds)
@@ -520,13 +520,11 @@ func (d *DispatcherImpl) handleTaskBulkAssignedTask(ctx context.Context, msg *ms
 			continue
 		}
 
-		for workerId, ids := range innerMsg.WorkerIdToTaskIds {
-			if len(ids) == 0 {
-				continue
-			}
+		for workerId, taskIds := range innerMsg.WorkerIdToTaskIds {
+			workerId := workerId
 
 			outerEg.Go(func() error {
-				return d.sendTasksToWorker(ctx, requeue, msg.TenantID, workerId, ids, taskIdToData)
+				return d.sendTasksToWorker(ctx, requeue, msg.TenantID, workerId, taskIds, taskIdToData)
 			})
 		}
 	}
@@ -680,12 +678,7 @@ func (d *DispatcherImpl) populateTaskData(
 		}
 	}
 
-	bulkV1Tasks := make([]*sqlcv1.V1Task, len(bulkDatas))
-	for i, task := range bulkDatas {
-		bulkV1Tasks[i] = task
-	}
-
-	parentDataMap, err := d.repov1.Tasks().ListTaskParentOutputs(ctx, tenantId, bulkV1Tasks)
+	parentDataMap, err := d.repov1.Tasks().ListTaskParentOutputs(ctx, tenantId, bulkDatas)
 
 	if err != nil {
 		for _, task := range bulkDatas {
@@ -1144,10 +1137,7 @@ func (d *DispatcherImpl) handleTaskCancelled(ctx context.Context, msg *msgqueue.
 		task, ok := taskIdsToTasks[msg.TaskId]
 
 		if !ok {
-			d.l.Warn().Ctx(ctx).Msgf("task %d not found in retry counts", msg.TaskId)
-			continue
-		}
-		if task == nil {
+			d.l.Warn().Ctx(ctx).Msgf("task %d not found", msg.TaskId)
 			continue
 		}
 

…0000_v1_0_116_batch_broadcast_output.sql …0000_v1_0_118_batch_broadcast_output.sqlcmd/hatchet-migrate/migrate/migrations/20260608000000_v1_0_116_batch_broadcast_output.sql renamed to cmd/hatchet-migrate/migrate/migrations/20260612000000_v1_0_118_batch_broadcast_output.sql cmd/hatchet-migrate/migrate/migrations/20260608000000_v1_0_116_batch_broadcast_output.sql renamed to cmd/hatchet-migrate/migrate/migrations/20260612000000_v1_0_118_batch_broadcast_output.sql

@@ -272,20 +272,9 @@ SELECT
     t.concurrency_strategy_ids,
     t.concurrency_keys,
     t.retry_backoff_factor,
-    t.retry_max_backoff,
-    tr.batch_id AS runtime_batch_id,
-    tr.batch_size AS runtime_batch_size,
-    tr.batch_index AS runtime_batch_index,
-    tr.worker_id AS runtime_worker_id,
-    tr.timeout_at AS runtime_timeout_at
+    t.retry_max_backoff
 FROM
     v1_task t
-LEFT JOIN
-    v1_task_runtime tr ON tr.task_id = t.id
-    AND tr.task_inserted_at = t.inserted_at
-    AND tr.retry_count = t.retry_count
-    AND tr.tenant_id = t.tenant_id
-    AND t.batch_key IS NOT NULL
 WHERE
     t.tenant_id = @tenantId::uuid
     AND t.id = ANY(@ids::bigint[]);

internal/services/dispatcher/dispatcher.go

@@ -127,12 +127,6 @@ type HatchetContext interface {
 	TriggeringEventId() *string
 
 	TriggeringEventKey() *string
-
-	// BatchId returns the batch ID if this step run is part of a batch task, or nil otherwise.
-	BatchId() *string
-
-	// BatchIndex returns the index of this item within the batch, or nil if not a batch item.
-	BatchIndex() *int32
 }
 
 // DurableEvictionHook observes waiting state transitions for a durable run so an
@@ -418,14 +412,6 @@ func (h *hatchetContext) DurableTaskInvocationCount() int32 {
 	return 0
 }
 
-func (h *hatchetContext) BatchId() *string {
-	return h.a.BatchId
-}
-
-func (h *hatchetContext) BatchIndex() *int32 {
-	return h.a.BatchIndex
-}
-
 func (h *hatchetContext) SetDurableEvictionHook(hook DurableEvictionHook) {
 	h.evictionHookMu.Lock()
 	defer h.evictionHookMu.Unlock()

pkg/repository/sqlcv1/tasks.sql

@@ -182,14 +182,6 @@ func (c *testHatchetContext) WorkerId() string {
 	panic("not implemented")
 }
 
-func (c *testHatchetContext) BatchId() *string {
-	return nil
-}
-
-func (c *testHatchetContext) BatchIndex() *int32 {
-	return nil
-}
-
 func TestAddMiddleware(t *testing.T) {
 	m := middlewares{}
 	middlewareFunc := func(ctx HatchetContext, next func(HatchetContext) error) error {

pkg/repository/sqlcv1/tasks.sql.go

@@ -141,12 +141,6 @@ type Worker struct {
 	id *string
 
 	panicHandler func(ctx HatchetContext, recovered any)
-
-	// batchFns maps actionId → BatchWrappedFn for batch task processing.
-	batchFns sync.Map
-
-	// batchStates maps batchId → *batchGroupState for in-flight batch tracking.
-	batchStates sync.Map
 }
 
 // Deprecated: WorkerOpt is an internal type used by the new Go SDK.
@@ -467,17 +461,6 @@ func (w *Worker) On(t triggerConverter, workflow workflowConverter) error {
 	return svc.(*Service).On(t, workflow)
 }
 
-// RegisterBatchAction registers a batch action function for the given action ID.
-// When a batch step run arrives, the worker buffers it and calls fn with all items once the batch is ready.
-func (w *Worker) RegisterBatchAction(actionId string, fn BatchWrappedFn) error {
-	w.batchFns.Store(actionId, fn)
-
-	// Register a regular action whose sole job is to enqueue this item and wait for the batch result.
-	return w.RegisterAction(actionId, func(ctx HatchetContext) (interface{}, error) {
-		return w.handleBatchItem(ctx, actionId)
-	})
-}
-
 // Deprecated: RegisterAction is an internal method used by the new Go SDK.
 // Use the new Go SDK at github.com/hatchet-dev/hatchet/sdks/go instead of calling this directly. Migration guide: https://docs.hatchet.run/home/migration-guide-go
 //
@@ -679,12 +662,6 @@ func (w *Worker) executeAction(ctx context.Context, assignedAction *client.Actio
 		return w.cancelStepRun(ctx, assignedAction)
 	case client.ActionTypeStartGetGroupKey:
 		return w.startGetGroupKey(ctx, assignedAction)
-	case client.ActionTypeStartBatch:
-		if assignedAction.BatchId == nil || assignedAction.BatchStart == nil {
-			return fmt.Errorf("START_BATCH action missing batch ID or batch start payload")
-		}
-		w.handleStartBatch(assignedAction.ActionId, *assignedAction.BatchId, assignedAction.BatchStart.ExpectedSize)
-		return nil
 	default:
 		return fmt.Errorf("unknown action type: %s", assignedAction.ActionType)
 	}