Skip to content

Successful RQ job execution does not get removed from StartedJobRegistry #692

Description

@choo0166

I have defined a job using the job decorator syntax like:

@job("calcium", timeout=3600, on_success=report_success, on_failure=report_failure)
def my_task(payload):
    start = time.perf_counter()
    response = requests.post(
        url,
        headers={
            "Content-Type": "application/json",
        },
        json=payload,
    )
    response.raise_for_status()
    end = time.perf_counter()

    response_body = response.json()

    return response_body

I have also defined the success callback handler:

def report_success(job, connection, result, *args, **kwargs):
    try:
        analysis = Analysis.objects.get(id=job.id)
        analysis.status = Analysis.Status.COMPLETED
        if "prediction" in result:
            PredictionResult.objects.create(
                analysis=analysis, prediction=result["prediction"]
            )
        if "segmentation" in result:
            # handle decoding and storing of base64 encoded segmentation masks
            for tp, (name, data) in result["segmentation"].items():
                # decode base64 string into binary and save to filefield
                f = ContentFile(content=base64.b64decode(data), name=name)
                SegmentationResult.objects.create(analysis=analysis, segmentation_mask=f, mask_type=tp)
        if "artifacts" in result:
            # handle decoding and storing of intermediate model artifacts
            for tp, (name, data) in result["artifacts"].items():
                f = ContentFile(content=base64.b64decode(data), name=name)
                AnalysisArtifact.objects.create(analysis=analysis, artifact=f, artifact_type=tp)
        analysis.save()
        logger.info(f"Analysis {analysis.id} processed successfully")
    except Analysis.DoesNotExist:
        logger.error(f"Analysis with job id {job.id} not found")

Basically, it handles saving of base64 encoded data into the respective Django model filefields.

When the job execution completes, RQ fails to remove the job from the StartedJobRegistry:

ai_worker-1     | Analysis 8df26860-bdcc-49da-a537-43088e7bddca processed successfully
ai_worker-1     | Handling successful execution of job 8df26860-bdcc-49da-a537-43088e7bddca
ai_worker-1     | Saving job 8df26860-bdcc-49da-a537-43088e7bddca's successful execution result
ai_worker-1     | Removing job 8df26860-bdcc-49da-a537-43088e7bddca from StartedJobRegistry
ai_worker-1     | Saving job 8df26860-bdcc-49da-a537-43088e7bddca's successful execution result
ai_worker-1     | Sent heartbeat to prevent worker timeout. Next one should arrive in 90 seconds.
ai_worker-1     | Removing job 8df26860-bdcc-49da-a537-43088e7bddca from StartedJobRegistry
ai_worker-1     | Saving job 8df26860-bdcc-49da-a537-43088e7bddca's successful execution result

The last two log lines basically repeat themselves till the job timeout is reached.

Any ideas to debug this issue?

Metadata

Metadata

Assignees

No one assigned

    Labels

    No labels
    No labels

    Type

    No type

    Fields

    No fields configured for issues without a type.

    Projects

    No projects

    Milestone

    No milestone

    Relationships

    None yet

    Development

    No branches or pull requests

    Issue actions