## Bug description

async migrations celery task id will always be "" through the migration

## Expected behavior

It will be set

## How to reproduce

1. Start a migration

## Environment

- [x] PostHog Cloud
- [ ] self-hosted PostHog (ClickHouse-based), version/commit: _please provide_
- [ ] self-hosted PostHog (Postgres-based, legacy), version/commit: _please provide_

## Additional context

The cause is a bad state machine: We set it in #trigger_migration but #mark_async_migration_as_running later in the chain always resets it.

cc @tiina303 and @yakkomajuri for prioritization

#### *Thank you* for your bug report – we love squashing them!

posthog/async_migrations/utils.py
"""
    In theory this is dangerous, as it can cause another task to be lost 
    `revoke` with `terminate=True` kills the process that's working on the task
    and there's no guarantee the task will not already be done by the time this happens.
    See: https://docs.celeryproject.org/en/stable/reference/celery.app.control.html#celery.app.control.Control.revoke
    However, this is generally ok for us because:
    1. Given these are long-running migrations, it is statistically unlikely it will complete during in between 
    this call and the time the process is killed
    2. Our Celery tasks are not essential for the functioning of PostHog, meaning losing a task is not the end of the world
    """
    app.control.revoke(migration_instance.celery_task_id, terminate=True)
    process_error(migration_instance, error)
def rollback_migration(migration_instance: AsyncMigration, force: bool = False):
    from posthog.async_migrations.runner import attempt_migration_rollback
    attempt_migration_rollback(migration_instance, force=force)
def complete_migration(migration_instance: AsyncMigration):
    update_async_migration(
        migration_instance=migration_instance,
        status=MigrationStatus.CompletedSuccessfully,
        finished_at=datetime.now(),
        progress=100,
    )
    from posthog.async_migrations.runner import run_next_migration
    next_migration = DEPENDENCY_TO_ASYNC_MIGRATION.get(migration_instance.name)
    if next_migration:
        run_next_migration(next_migration)
def mark_async_migration_as_running(migration_instance: AsyncMigration):