services/celery_data_processing.py
def create_file_processing_tasks():
    """ Generates tasks to enqueue.  This is called every 6 minutes, and tasks have a lifetime
    of 6 minutes.  Note that tasks are not removed from the queue by RabbitMQ, but by Celery.
    inspecting the queue will continue to display the tasks that have not been sent to Celery
    until the most recent job is finished.

    Also, for some reason 5 minutes is the smallest value that .... works.  At all.
    No clue why.
    """

    # set the tasks to expire at the 5 minutes and thirty seconds mark after the most recent
    # 6 minutely cron task. This way all tasks will be revoked at the same, and well-known, instant.
    expiry = (datetime.utcnow() + timedelta(minutes=5)).replace(second=30, microsecond=0)

    with make_error_sentry(sentry_type=SentryTypes.data_processing):
        participant_set = set(
            Participant.objects.filter(files_to_process__isnull=False)
                .distinct()
                # .order_by("id")  # For debugging, forces overlap conflicts.
                .order_by("?")     # don't want a single user blocking everyone because they are at the front.
                .values_list("id", flat=True)
        )
        
        # sometimes celery just fails to exist, set should be redundant.
        active_set = set(get_processing_active_job_ids())
        
        participants_to_process = participant_set - active_set
        print("Queueing these participants:", ",".join(str(p) for p in participants_to_process))

        for participant_id in participants_to_process:
            # Queue all users' file processing, and generate a list of currently running jobs
            # to use to detect when all jobs are finished running.
            safe_apply_async(
                celery_process_file_chunks,
                args=[participant_id],
                max_retries=0,
                expires=expiry,
                task_track_started=True,
                task_publish_retry=False,
                retry=False
            )
        print(f"{len(participants_to_process)} users queued for processing")
 Generates tasks to enqueue.  This is called every 6 minutes, and tasks have a lifetime
    of 6 minutes.  Note that tasks are not removed from the queue by RabbitMQ, but by Celery.
    inspecting the queue will continue to display the tasks that have not been sent to Celery
    until the most recent job is finished.

    Also, for some reason 5 minutes is the smallest value that .... works.  At all.
    No clue why.
    
Generates tasks to enqueue.  This is called every 6 minutes, and tasks have a lifetime
of 6 minutes.  Note that tasks are not removed from the queue by RabbitMQ, but by Celery.
inspecting the queue will continue to display the tasks that have not been sent to Celery
until the most recent job is finished.

Also, for some reason 5 minutes is the smallest value that .... works.  At all.
No clue why.
Generates tasks to enqueue . This is called every 6 minutes and tasks have a lifetime of 6 minutes . Note that tasks are not removed from the queue by RabbitMQ but by Celery . inspecting the queue will continue to display the tasks that have not been sent to Celery until the most recent job is finished . Also for some reason 5 minutes is the smallest value that .... works . At all . No clue why .
def create_file_processing_tasks():
    expiry = (datetime.utcnow() + timedelta(minutes=5)).replace(second=30, microsecond=0)
    with make_error_sentry(sentry_type=SentryTypes.data_processing):
        participant_set = set(
            Participant.objects.filter(files_to_process__isnull=False)
                .distinct()
                .order_by("?")     
                .values_list("id", flat=True)
        )
        active_set = set(get_processing_active_job_ids())
        participants_to_process = participant_set - active_set
        print("Queueing these participants:", ",".join(str(p) for p in participants_to_process))
        for participant_id in participants_to_process:
            safe_apply_async(
                celery_process_file_chunks,
                args=[participant_id],
                max_retries=0,
                expires=expiry,
                task_track_started=True,
                task_publish_retry=False,
                retry=False
            )
        print(f"{len(participants_to_process)} users queued for processing")
def create_file_processing_tasks ( ) : expiry = ( datetime . utcnow ( ) + timedelta ( minutes = 5 ) ) . replace ( second = 30 , microsecond = 0 ) with make_error_sentry ( sentry_type = SentryTypes . data_processing ) : participant_set = set ( Participant . objects . filter ( files_to_process__isnull = False ) . distinct ( ) . order_by ( "?" ) . values_list ( "id" , flat = True ) ) active_set = set ( get_processing_active_job_ids ( ) ) participants_to_process = participant_set - active_set print ( "Queueing these participants:" , "," . join ( str ( p ) for p in participants_to_process ) ) for participant_id in participants_to_process : safe_apply_async ( celery_process_file_chunks , args = [ participant_id ] , max_retries = 0 , expires = expiry , task_track_started = True , task_publish_retry = False , retry = False ) print ( f"{len(participants_to_process)} users queued for processing" )
Generates tasks to enqueue.
Generates tasks to enqueue .
""" Generates tasks to enqueue.  This is called every 6 minutes, and tasks have a lifetime
    of 6 minutes.  Note that tasks are not removed from the queue by RabbitMQ, but by Celery.
    inspecting the queue will continue to display the tasks that have not been sent to Celery
    until the most recent job is finished.

    Also, for some reason 5 minutes is the smallest value that .... works.  At all.
    No clue why.
    """ # set the tasks to expire at the 5 minutes and thirty seconds mark after the most recent # 6 minutely cron task. This way all tasks will be revoked at the same, and well-known, instant. # .order_by("id")  # For debugging, forces overlap conflicts. # don't want a single user blocking everyone because they are at the front. # sometimes celery just fails to exist, set should be redundant. # Queue all users' file processing, and generate a list of currently running jobs # to use to detect when all jobs are finished running.