Building async batch processors for cold chain data lakes

from datetime import datetime, timezone
from enum import Enum
from typing import Optional
from pydantic import BaseModel, Field, ValidationError, ConfigDict

class ComplianceState(str, Enum):
    WITHIN_SPEC = "within_spec"
    EXCURSION = "excursion"
    UNKNOWN = "unknown"

class TelemetryRecord(BaseModel):
    model_config = ConfigDict(frozen=True, extra="forbid")
    
    sensor_id: str = Field(..., min_length=4, max_length=64, description="Unique hardware identifier")
    timestamp_utc: datetime = Field(..., description="Original sensor-generated UTC timestamp")
    temperature_c: float = Field(..., ge=-100.0, le=100.0, description="Temperature in Celsius")
    zone_id: str = Field(..., pattern=r"^[A-Z0-9\-]+$", description="Cold storage zone identifier")
    compliance_state: ComplianceState = ComplianceState.UNKNOWN
    raw_payload_hash: Optional[str] = None

    @classmethod
    def from_raw(cls, payload: dict) -> tuple[Optional["TelemetryRecord"], Optional[str]]:
        """Validates payload and returns ``(record, None)`` or ``(None, error_msg)``."""
        try:
            record = cls(**payload)
            return record, None
        except ValidationError as e:
            return None, str(e)

import asyncio
import hashlib
import json
import logging
from collections import deque
from datetime import datetime, timezone
from typing import List, Dict, Any, Optional

logger = logging.getLogger(__name__)

class AsyncColdChainBatcher:
    def __init__(self, max_batch_size: int = 5000, flush_interval_sec: float = 10.0, s3_client=None):
        self.max_batch_size = max_batch_size
        self.flush_interval_sec = flush_interval_sec
        self.s3_client = s3_client
        self._buffer: deque = deque()
        self._lock = asyncio.Lock()
        self._flush_event = asyncio.Event()
        self._running = False
        self._task: Optional[asyncio.Task] = None
        self._dead_letter_queue: List[Dict[str, Any]] = []
        self._dlq_lock = asyncio.Lock()

    async def start(self):
        self._running = True
        self._task = asyncio.create_task(self._periodic_flush_loop())
        logger.info("Async batch processor started")

    async def stop(self):
        self._running = False
        if self._task:
            self._task.cancel()
            try:
                await self._task
            except asyncio.CancelledError:
                pass
        await self._flush_batch(force=True)
        logger.info("Async batch processor stopped")

    async def ingest(self, raw_payload: Dict[str, Any]):
        record, error = TelemetryRecord.from_raw(raw_payload)
        if record:
            async with self._lock:
                self._buffer.append(record)
                if len(self._buffer) >= self.max_batch_size:
                    self._flush_event.set()
        else:
            async with self._dlq_lock:
                self._dead_letter_queue.append({
                    "raw": raw_payload,
                    "validation_error": error,
                    "rejected_at_utc": datetime.now(timezone.utc).isoformat()
                })
                if len(self._dead_letter_queue) >= 1000:
                    await self._flush_dlq()

    async def _periodic_flush_loop(self):
        while self._running:
            try:
                await asyncio.wait_for(self._flush_event.wait(), timeout=self.flush_interval_sec)
            except asyncio.TimeoutError:
                pass
            except asyncio.CancelledError:
                raise
            try:
                await self._flush_batch()
            except Exception:
                # Never let a single flush failure kill the periodic loop; the
                # error is already logged inside _write_to_datalake and the
                # records are re-buffered for retry.
                logger.exception("Periodic flush cycle failed; loop continues")
            finally:
                # Clear the event AFTER the flush so any set() that arrived
                # during the flush triggers another cycle immediately.
                async with self._lock:
                    if not self._buffer:
                        self._flush_event.clear()

    async def _flush_batch(self, force: bool = False):
        async with self._lock:
            if not self._buffer:
                return
            batch_records = list(self._buffer)
            self._buffer.clear()

        # Compute audit digest over canonical JSON
        canonical_json = json.dumps(
            [r.model_dump(mode="json") for r in batch_records],
            sort_keys=True, separators=(",", ":"),
        )
        batch_digest = hashlib.sha256(canonical_json.encode("utf-8")).hexdigest()

        # Partition by each record's OWN timestamp_utc, not flush wall-clock,
        # so late-arriving telemetry lands in the partition the data belongs
        # to. We pick the median to bias toward the bulk of records.
        timestamps = sorted(r.timestamp_utc for r in batch_records)
        midpoint = timestamps[len(timestamps) // 2]
        partition_key = f"year={midpoint:%Y}/month={midpoint:%m}/day={midpoint:%d}"
        object_key = f"coldchain/{partition_key}/batch_{batch_digest}.json"

        await self._write_to_datalake(object_key, batch_records, batch_digest)

    async def _write_to_datalake(self, key: str, records: List[TelemetryRecord], digest: str):
        if not self.s3_client:
            logger.info("[SIMULATION] Flushing %d records to %s | Digest: %s", len(records), key, digest)
            return
        try:
            payload = json.dumps({
                "metadata": {
                    "batch_digest": digest,
                    "record_count": len(records),
                    "flush_utc": datetime.now(timezone.utc).isoformat(),
                },
                "records": [r.model_dump(mode="json") for r in records],
            }).encode("utf-8")
            await self.s3_client.put_object(
                Bucket="pharma-coldchain-datalake",
                Key=key,
                Body=payload,
            )
            logger.info("Successfully flushed batch to %s", key)
        except Exception as e:
            # Never silently drop a batch — ALCOA+ Complete demands every
            # validated record reaches durable storage. Re-buffer at the
            # head of the deque and let the next cycle retry. Operators
            # should bound retries by monitoring deque depth.
            logger.error("Data lake write failed for %s: %s; re-buffering batch", key, e)
            async with self._lock:
                self._buffer.extendleft(reversed(records))
                self._flush_event.set()
            raise

    async def _flush_dlq(self):
        async with self._dlq_lock:
            if not self._dead_letter_queue:
                return
            dlq_dump = self._dead_letter_queue.copy()
            self._dead_letter_queue.clear()
        logger.warning(f"Flushing {len(dlq_dump)} invalid records to dead-letter storage")
        # Implement DLQ persistence logic here (e.g., separate S3 bucket or Kafka topic)

import aioboto3
import asyncio

async def run_processor():
    async with aioboto3.Session().client("s3") as s3:
        processor = AsyncColdChainBatcher(
            max_batch_size=2000,
            flush_interval_sec=15.0,
            s3_client=s3
        )
        await processor.start()
        
        # Simulate high-throughput ingestion
        for i in range(10000):
            await processor.ingest({
                "sensor_id": f"CC-SENSOR-{i % 50:03d}",
                "timestamp_utc": datetime.now(timezone.utc).isoformat(),
                "temperature_c": 4.2 + (i % 10) * 0.1,
                "zone_id": "FRZ-A1",
                "compliance_state": "within_spec"
            })
            if i % 500 == 0:
                await asyncio.sleep(0.01) # Yield to event loop
                
        await processor.stop()

if __name__ == "__main__":
    logging.basicConfig(level=logging.INFO)
    asyncio.run(run_processor())