Logistics event service
Fern Logistics needs a local event-processing component. The input contains job records, invalid values, duplicates, supplied handler failures and a shutdown marker. Deliver ordered completed jobs, a recoverable error list and a bounded intake path that drains before returning.
Fern Logistics receives a mixture of parcel jobs, control events and retried deliveries. The service must process valid work once, record rejected jobs and finish accepted work during shutdown. A loop that handles only the successful example will lose those distinctions.
You grow one service function through six checked changes. Earlier steps use awaited handlers; the finished version connects a producer and consumer through a bounded asyncio queue. The records are supplied in memory, but the coroutines and queue run in the browser. Every accepted queue item needs completion accounting, including the sentinel. Keep duplicate handling ahead of effects so a failed first delivery cannot quietly become a different result when the same ID appears again.
import json
import asyncio
async def process_payload(payload: str) -> str:
data = json.loads(payload)
done, errors, seen = [], [], set()
async def handle(event):
await asyncio.sleep(0)
done.append(event['id'] + ':' + str(event['value'] * 2))
for event in data['events']:
await handle(event)
return json.dumps({'done':done,'errors':errors},sort_keys=True,separators=(',',':'))
An awaited handler finishes before its result is reported, including zero values. Keep intake, validation and the owned effect boundary separate. Await handlers so accepted work finishes before returning. The service visits each accepted event once; the final queue bounds pending intake and preserves FIFO handling.