Cursor is not deserialized correctly with multiple columns

[nigel-nava]

The bug:
Pausing and resuming from the dashboard when my task has multiple cursor columns fails with an error.

The error:

ActiveRecord::StatementInvalid
PG::InvalidDatetimeFormat: ERROR: invalid input syntax for type timestamp with time zone: "["2025-03-11 14:30:01.132797000", "7d826fb4-b75e-4393-a799-e8cacbb32624"]" CONTEXT: unnamed portal parameter $1 = '...' 

The code:

# base jobclass
class DataMigration::MaintenanceTasksBaseJob < MaintenanceTasks::TaskJob
  queue_as :latency_30s
end

class ExampleTask < MaintenanceTasks::Task
  def collection
    User.all # has "id" and "created_at" columns
  end

  def cursor_columns
    [ :created_at, :id ]
  end

  def process(user)
    puts "Processing #{user.id}, #{user.created_at}"
    sleep 1
  end
end

My fix:

# base job class
class DataMigration::MaintenanceTasksBaseJob < MaintenanceTasks::TaskJob
  queue_as :latency_30s

  private

  def build_enumerator(run, cursor:)
    deserialized_cursor = cursor
    if cursor.nil? && @task.cursor_columns.length > 1 && run.cursor.is_a?(String)
      deserialized_cursor = JSON.parse(run.cursor)
    end
    super(run, cursor: deserialized_cursor)
  end
end

I think what's happening is that pausing and resuming from the dashboard enqueues a new job with no serialized position. In ac3872a we assign the cursor_position to the value of run.cursor, but at this point the value is supposed to be fully deserialized. When there are multiple columns in cursor_columns, we actually have a JSON string, which is not handled correctly by JobIteration

[github-actions]

This issue has been marked as stale because it has not been commented on in two months.
Please reply in order to keep the issue open. Otherwise, it will close in 14 days.
Thank you for contributing!

[github-actions]

This issue has been marked as stale because it has not been commented on in two months.
Please reply in order to keep the issue open. Otherwise, it will close in 14 days.
Thank you for contributing!

[ClaytonPassmore]

This is tricky, because if the relation has a composite primary key and no cursor_columns are specified on the task, you'll still end up with a complex cursor value but be none the wiser.

The root of the problem is that the cursor being passed in to the enumerator is not in the format it was yielded in. Consistently encoding the yielded cursor as a JSON object (and thus deserializing it on resume) is probably the easiest solution. The challenge will be adding support for that without breaking existing cursors in the wild.

While it wouldn't be the most elegant, you could add a flag to the maintenance_tasks_runs table that indicates whether the column is JSON (or whether it's the old format). E.g.

def build_enumerator(run, cursor:)
  cursor = JSON.parse(run.cursor) if cursor.nil? && run.cursor? && run.cursor_is_json?
  cursor ||= run.cursor
  # ..

New run records can default to having cursor_is_json: true.

If this solution is reasonable, I can take a stab at submitting a pull request, let me know.