Bulk Ingestion Architecture: The S3 Token-Pointer Pattern
// How we process 10,000 property records in under 20 seconds without choking SQS message limits
Enterprise real estate acquisition pipelines ingest massive batches of county assessor data, pre-foreclosure filings, and skip-traced lead sheets. In high-volume operations, a single daily upload can contain 10,000 to 50,000 property records, easily totaling tens of megabytes of raw text and metadata.
A common architectural blunder in serverless pipelines is attempting to push bulk records directly through message queues like Amazon SQS or EventBridge. SQS enforces a strict hard limit of 256 KB per message payload. Even if records are split into micro-messages, transmitting thousands of individual SQS messages saturates queue throughput, triggers heavy network serialization costs, and inflates AWS transfer latency.
To overcome this, we implemented the S3 Token-Pointer Pattern. When an operator or automated webhook uploads a raw list, the ingestion worker writes the raw payload directly to Amazon S3, generates a unique batch ticket, and publishes a lightweight JSON pointer token (under 1 KB) to the event bus. Background processing workers sized at 1024 MB RAM (0.58 vCPU, 60s timeout) read the token, stream the specific S3 chunk into memory, and screen 10,000 properties against custom investment criteria in under 20 seconds.
- ▪Zero SQS payload bottlenecks: Event buses only carry lightweight pointer metadata (S3 bucket, object key, partition range, and batch ID).
- ▪Sub-20-second evaluation: Ingests, normalizes, and screens 10,000 property records against equity and location criteria before any database write.
- ▪Cost and bandwidth optimization: Workers stream only the exact byte range needed, bypassing redundant serialization round-trips.
from pydantic import BaseModel, Field
class LeadBatchIngestionToken(BaseModel):
"""
Lightweight pointer token passed across SQS / EventBridge.
Eliminates 256 KB payload limits by referencing immutable S3 batch storage.
"""
batch_id: str = Field(description="Unique UUID for the ingestion run")
s3_bucket: str = Field(description="Target storage bucket")
s3_key: str = Field(description="Path to compressed JSON/CSV file in S3")
total_records: int = Field(gt=0, description="Record count within the batch")
campaign_id: str
target_region: str = Field(description="Two-letter state code, e.g., 'FL', 'TX', 'CA'")
checksum_sha256: str = Field(description="Integrity checksum verified before parsing")Data Sanitization & Phone Normalization in parser.py
// Cleaning raw county assessor sheets, enforcing E.164 formatting, and handling multi-number owner profiles
Public property records and skip-trace data are notoriously noisy. Phone numbers appear in dozens of arbitrary formats, such as (555) 019-2834, 555.019.2834, +1-555-019-2834, or raw 10-digit strings with missing country prefixes. Furthermore, assessor records frequently list multiple phone numbers for a single owner, mixing active mobile numbers with disconnected landlines.
In our microservice topology, data sanitization is strictly isolated inside parser.py before domain logic in executor.py ever touches a record. The parser runs regular expression pre-filtering followed by strict E.164 standardization (+1XXXXXXXXXX). Records with invalid length, test prefixes (like 555-0100 to 555-0199), or unparseable area codes are rejected at the gate and logged to a validation audit stream.
For owners with multiple contact numbers, the parser prioritizes phone line types based on carrier metadata. Mobile lines are flagged for conversational voice and SMS capability, while verified landlines are restricted to voice-only outreach. Duplicate numbers across distinct property parcels are consolidated under a master owner entity to prevent the system from dialing the same seller multiple times for separate properties.
- ▪Strict E.164 compliance: All numbers are standardized to international +1 format before entering the database or telephony bridge.
- ▪Multi-number hierarchy: Prioritizes verified mobile lines over landlines and filters out non-routable carrier numbers.
- ▪Cross-parcel deduplication: Maps multiple properties to a single owner profile, guaranteeing a seller is never contacted twice for adjacent lots.
Geographic Caller Routing & TCPA Safe-Hours Enforcement
// Why a single-concurrency FIFO metronome is mathematically required to eliminate carrier spam flags and prevent illegal calling windows
Under the United States Telephone Consumer Protection Act (TCPA) and mini-TCPA state statutes (such as Florida FTSA and Oklahoma regulations), outbound automated calling is strictly restricted to recipient local hours, typically between 8:00 AM and 8:00 PM. Calling a prospect at 7:45 AM or after 8:00 PM local time carries statutory penalties of up to $1,500 per call.
The technical challenge lies in timezone determination. A phone number area code does not guarantee physical location because consumers routinely keep mobile numbers when moving across time zones. Our system combines a dual-lookup strategy: it cross-references the phone number area code with the physical property county GIS timezone and enforces the more restrictive time window of the two.
To mathematically guarantee compliance, the Pacing Metronome runs as a dedicated single-concurrency FIFO worker (concurrency = 1, FIFO locked). If a batch contains leads across Eastern (EST) and Pacific (PST) time zones, the metronome evaluates the recipient local clock before dispatching. Leads outside the legal window are deferred to an SQS delay partition or held until 8:00 AM local time, preventing accidental dispatch during forbidden hours.
- ▪Dual-coordinate timezone resolution: Cross-references phone area codes with property county GIS data, always applying the most restrictive safe window.
- ▪Single-concurrency FIFO lock: Guarantees sequential, deterministic pacing without clock drift or race conditions between parallel dispatch workers.
- ▪Protection against carrier spam labeling: Steady, metronomic call dispatching prevents sudden volumetric spikes that trigger carrier STIR/SHAKEN flags.
The 3-Tier Waterfall Configuration Engine: Fail-Safe Resilience
// How microservices survive database pool saturation and SSM control-plane outages without crashing or violating regulations
In an automated voice platform, microservices require continuous access to operational settings: current dialing rates, tenant concurrency quotas, emergency kill switches, and regional safe-hour rules. If a service cannot fetch its configuration, it risks either crashing (dropping live calls) or failing open (dialing without limits at 3:00 AM).
To solve this, we engineered a 3-tier waterfall configuration fallback: Tier 1 queries dynamic profiles directly from PostgreSQL via RDS Proxy; Tier 2 falls back to AWS Systems Manager (SSM Parameter Store); and Tier 3 loads immutable, ultra-conservative local defaults packaged directly within the microservice repository.
Under normal conditions, operators adjust pacing live in PostgreSQL with zero deployment downtime. If the database experiences pool exhaustion or failover latency, the service transparently retrieves cached parameters from AWS Systems Manager. If AWS SSM itself is throttled during an AWS control-plane event, the service defaults to Tier 3 local constants. Crucially, Tier 3 defaults to the safest possible parameters (minimum rate, shortest calling window), ensuring the system never defaults to an aggressive or non-compliant state.
- ▪Tier 1 (PostgreSQL Dynamic): Operators modify dialing pacing and campaign parameters live via SQL or admin dashboards with zero downtime.
- ▪Tier 2 (AWS Systems Manager): Transparently absorbs database connection spikes or pool exhaustion with cached cloud parameters.
- ▪Tier 3 (Local Hardened Defaults): Immutable local fallback guaranteeing fail-safe, ultra-conservative operation during total upstream outages.
import logging
from typing import NamedTuple
from core.database.connection import get_db_cursor
from core.aws.ssm import get_cached_ssm_parameter
logger = logging.getLogger("wholesale.config")
class OperationalProfile(NamedTuple):
calls_per_second: float
max_concurrency: int
safe_window_start_hour: int
safe_window_end_hour: int
source_tier: str
# Tier 3: Immutable local fail-safe defaults (strictly conservative)
LOCAL_FAILSAFE_PROFILE = OperationalProfile(
calls_per_second=0.5, # Ultra-conservative 1 call every 2 seconds
max_concurrency=5, # Minimal concurrency
safe_window_start_hour=9, # Conservative 9:00 AM (TCPA allows 8:00 AM)
safe_window_end_hour=17, # Conservative 5:00 PM (TCPA allows 8:00 PM)
source_tier="TIER_3_LOCAL_EMERGENCY"
)
def resolve_operational_profile(region_code: str) -> OperationalProfile:
"""
Cascades through 3 configuration tiers to guarantee zero downtime
and prevent illegal or unthrottled outbound dialing.
"""
# Tier 1: Primary PostgreSQL dynamic table
try:
with get_db_cursor(read_only=True, timeout_ms=800) as cursor:
cursor.execute("SELECT cps, max_conc, start_hr, end_hr FROM campaign_profiles WHERE region = %s;", (region_code,))
row = cursor.fetchone()
if row:
return OperationalProfile(row[0], row[1], row[2], row[3], source_tier="TIER_1_DATABASE")
except Exception as db_err:
logger.warning(f"Tier 1 DB config fetch failed ({db_err}). Escalating to Tier 2 SSM.")
# Tier 2: AWS Systems Manager Parameter Store fallback
try:
ssm_data = get_cached_ssm_parameter(f"/wholesale/pacing/{region_code}", ttl_seconds=300)
if ssm_data:
return OperationalProfile(
calls_per_second=ssm_data["cps"],
max_concurrency=ssm_data["max_conc"],
safe_window_start_hour=ssm_data["start_hr"],
safe_window_end_hour=ssm_data["end_hr"],
source_tier="TIER_2_AWS_SSM"
)
except Exception as ssm_err:
logger.error(f"Tier 2 SSM fetch failed ({ssm_err}). Escalating to Tier 3 local defaults.")
# Tier 3: Local hardened microservice constants
logger.critical("Using Tier 3 emergency configuration profile. Outbound rate severely throttled for safety.")
return LOCAL_FAILSAFE_PROFILEThe Ghost-Lease Problem: DynamoDB TTL Stream Sweeper
// Automatically recovering stalled dialing slots and cleaning up zombie states when telecom carriers drop mid-call
In high-concurrency telephony, calls do not always end cleanly. A mobile user may drive into a cellular dead zone, a carrier trunk may crash, or an edge WebSocket connection may silently sever. When this happens, the telephony provider never sends a final completed webhook.
Without automated remediation, the dialing lease assigned to that contact remains locked in DynamoDB forever. Over time, these orphaned zombie leases accumulate, artificially exhausting tenant concurrency quotas and starving the outbound queue until the entire system grinds to a halt.
We resolved this with a DynamoDB Time-To-Live (TTL) stream sweeper. When a call begins, a lease record is created with an explicit expiration timestamp (current time + 6 minutes, which exceeds the maximum possible 5-minute call duration). If no clean completion webhook arrives before expiration, DynamoDB automatically deletes the item and emits an expiration event to DynamoDB Streams. A dedicated sweeper Lambda catches this stream event, releases the pacing slot, marks the lead as CALL_TIMED_OUT in PostgreSQL, and enqueues a retry ticket to SQS.
- ▪Automated zombie lease recovery: Stalled calls are reclaimed automatically by DynamoDB TTL without manual operator intervention.
- ▪Queue starvation prevention: Concurrency slots are continuously recycled even when telecom carriers crash mid-call.
- ▪Dead-letter audit trails: Unhandled exceptions or corrupt payloads are preserved in 14-day SQS DLQs for post-incident debugging.
Operational Takeaways for High-Volume Systems
// Hard-won engineering rules for building regulatory-compliant, failure-tolerant cloud telephony pipelines
Operating high-volume conversational AI telephony in production requires designing for inevitable external failures. Keep these four operational lessons in mind:
- ▪Never pass bulk payloads through event queues: Pass lightweight S3 pointer tokens and let workers stream specific slices into memory.
- ▪Enforce 3-tier waterfall configuration fallbacks: Always design systems to fail into ultra-conservative regulatory states rather than unthrottled defaults.
- ▪Lock outbound pacing to single-concurrency FIFO queues: Sequential dispatching eliminates race conditions and protects carrier reputation.
- ▪Rely on TTL-driven stream sweepers for lease cleanup: Assume that network carriers will drop 1% to 2% of calls without sending terminal status events.
Part 1: Distributed Architecture, Infrastructure, and Service Sizing
Designing or stabilizing a high-concurrency cloud telephony system?
We specialize in resilient cloud systems, real-time voice latency elimination, and bulletproof state synchronization under heavy production load.