Disconnect Between `build_service.py` and `test.py` with Redis and Celery Integration

[DoloMike]

Description

When running the project in WSL2, there is a disconnect between build_service.py and test.py due to the integration of Redis and Celery. Specifically:

1. test.py submits a task to Celery using the send_task method.
2. Celery enqueues the task in the Redis queue (minecraft_builder).
3. build_service.py picks up the task directly from Redis and processes it successfully.
4. However, Celery does not get updated about the task's completion since build_service.py is not a Celery worker and does not interact with Celery’s task tracking system.

This results in:

• The task status remaining PENDING when monitored using AsyncResult in test.py.
• Celery not being aware of task progress, completion, or results.

Steps to Reproduce

1. Start the Redis server.
2. Run python build_service.py to process tasks
3. In another terminal, run python test.py to submit and monitor a build task
4. Observe that:
   • The build job is processed successfully in build_service.py.
   • test.py continuously reports the task status as PENDING.

Root Cause

The issue arises because:

• test.py submits tasks to Celery, which enqueues them in Redis.
• build_service.py directly polls the Redis queue (minecraft_builder) and processes tasks independently, bypassing Celery's worker system.
• Celery expects a registered Celery worker to process tasks and update their status in its result backend (Redis), which does not happen here.

Fix Applied

Implemented a manual update to Celery's result backend in build_service.py to bridge the gap. Here's how I resolved it:

1. Imported Celery in build_service.py:

   from celery import Celery, states
   from celery.exceptions import Ignore
   celery_app = Celery('minecraft_builder', broker='redis://localhost:6379/0', backend='redis://localhost:6379/0')

3. Added a Helper Function to Update Task Status:

   def update_task_status(task_id, status, result=None, error=None):
       if status == states.SUCCESS:
           celery_app.backend.store_result(task_id, result, states.SUCCESS)
       elif status == states.FAILURE:
           celery_app.backend.store_result(task_id, error, states.FAILURE)
           raise Ignore()

4. Updated process_batch() in build_service.py:
   After processing each job, the task status was updated manually:

   for (job, result) in zip(jobs, results):
       if isinstance(result, Exception):
           logger.error(f"Job {job['id']} failed: {str(result)}")
           update_task_status(job['id'], states.FAILURE, error=str(result))
       else:
           logger.info(f"Job {job['id']} completed successfully")
           update_task_status(job['id'], states.SUCCESS, result={"status": "success"})

Outcome

After applying the fix:

• build_service.py processes tasks as before.
• test.py successfully monitors the task status, transitioning from PENDING to SUCCESS once the task is complete.
• The build result can now be retrieved via Celery.

Next Steps

For a long-term solution:

1. Consider converting build_service.py into a Celery worker to align with Celery's architecture. This eliminates the need for manual Redis polling and status updates.
2. Address related warnings and enhancements:
   • Missing structure_name and dimensions in build results.
   • MaxListenersExceededWarning in mineflayer.

Logs (Before Fix)

test.py Output:
Current status: PENDING (repeats indefinitely)

build_service.py Logs:
INFO - Retrieved 1 pending jobs from Redis
INFO - Job <task_id> completed successfully

Logs (After Fix)

test.py Output:
Current status: PENDING
Current status: SUCCESS

Build Result:
Status: success
Structure Name: None
Dimensions: None

build_service.py Logs:
INFO - Retrieved 1 pending jobs from Redis
INFO - Job <task_id> completed successfully
INFO - Updating task status to SUCCESS

Additional Context

This manual update resolves the issue temporarily but introduces additional maintenance overhead. Migrating to a Celery worker-based architecture is recommended for scalability and maintainability.