确认完成事件后 ACK 持久消息
This commit is contained in:
@@ -176,12 +176,20 @@ async def run_message(message: object, scheduler: OpenSandboxScheduler) -> None:
|
||||
runner_request = RunnerRequest.from_json(message.data)
|
||||
await scheduler.create(runner_request)
|
||||
task = scheduler.active[runner_request.job_id]
|
||||
while not task.done():
|
||||
try:
|
||||
await asyncio.wait_for(asyncio.shield(task), timeout=30)
|
||||
except asyncio.TimeoutError:
|
||||
await message.in_progress()
|
||||
await task
|
||||
try:
|
||||
while not task.done():
|
||||
try:
|
||||
await asyncio.wait_for(asyncio.shield(task), timeout=30)
|
||||
except asyncio.TimeoutError:
|
||||
await message.in_progress()
|
||||
await task
|
||||
except asyncio.CancelledError:
|
||||
# A completed webhook cancels the lifecycle monitor after its
|
||||
# Lifecycle DELETE succeeds. That is successful message handling.
|
||||
# During controller shutdown the scheduler still owns the job, so
|
||||
# preserve the unacked message for redelivery.
|
||||
if runner_request.job_id in scheduler.active:
|
||||
raise
|
||||
except ValueError as error:
|
||||
LOG.info("runner request already active: %s", error)
|
||||
await message.nak(delay=5)
|
||||
|
||||
Reference in New Issue
Block a user