fix: skip RetryWait for new tasks in scheduler (#459)

The scheduler was applying RetryWait to all tasks, including brand-new
ones with Retries=0. This caused every SendTransaction task to wait 5
minutes before being picked up, even though it had never been attempted.

The fix checks Retries==0 to identify new tasks and picks them up
immediately, only applying RetryWait to tasks that have actually failed
and are being retried.

Author: frrist

pkg/pdp/scheduler/engine.go

@@ -197,7 +197,8 @@ func (e *TaskEngine) pollerTryAllWork() bool {
 		var taskIDs []TaskID
 		for _, t := range tasks {
 			if h.TaskTypeDetails.RetryWait == nil ||
-				time.Since(t.UpdateTime) > h.TaskTypeDetails.RetryWait(0) {
+				t.Retries == 0 || // New task - pick up immediately
+				time.Since(t.UpdateTime) > h.TaskTypeDetails.RetryWait(int(t.Retries)) {
 				taskIDs = append(taskIDs, TaskID(t.ID))
 			}
 		}

pkg/pdp/scheduler/engine_test.go

@@ -789,3 +789,45 @@ func TestTaskEngineShutdownTimeoutMultipleTasks(t *testing.T) {
 	assert.Equal(t, numTasks, taskCount, "All tasks should have started")
 	taskCountMu.Unlock()
 }
+
+// TestTaskEngineNewTaskIgnoresRetryWait verifies that brand-new tasks (Retries=0)
+// are picked up immediately, ignoring the RetryWait duration. This is a regression
+// test for https://github.com/storacha/piri/issues/459
+func TestTaskEngineNewTaskIgnoresRetryWait(t *testing.T) {
+	db := setupTestDB(t)
+
+	taskExecuted := make(chan struct{})
+
+	mockTask := NewMockTask("test_task", true)
+	// Set a very long RetryWait - if applied to new tasks, test would timeout
+	mockTask.typeDetails.RetryWait = func(retries int) time.Duration {
+		return 5 * time.Minute
+	}
+	mockTask.doFunc = func(taskID scheduler.TaskID) (bool, error) {
+		close(taskExecuted)
+		return true, nil
+	}
+
+	engine, err := scheduler.NewEngine(db, []scheduler.TaskInterface{mockTask})
+	require.NoError(t, err)
+	err = engine.Start(t.Context())
+	require.NoError(t, err)
+	t.Cleanup(func() {
+		if err := engine.Stop(context.Background()); err != nil {
+			t.Logf("failed to stop engine: %v", err)
+		}
+	})
+
+	mockTask.WaitForReady()
+	mockTask.AddTask(func(tID scheduler.TaskID, tx *gorm.DB) (bool, error) {
+		return true, nil
+	})
+
+	// Task should be picked up within 5 seconds, NOT 5 minutes
+	select {
+	case <-taskExecuted:
+		// Success - task was picked up immediately
+	case <-time.After(5 * time.Second):
+		t.Fatal("Task should be picked up immediately for new tasks (Retries=0), but it wasn't")
+	}
+}