Multi-Tenancy and Isolation Questions

Requirements & constraints:

• Route each tenant’s queries to the tenant’s primary (writes, strong consistency) or an allowed region-replica (reads) while preserving per-tenant residency, low latency, and read-your-writes (RYW).
• Support fast cacheable routing decisions, failover, and multi-region replicas with replication-lag awareness.

High-level design:

1. Routing metadata

• Tenant record: tenant_id → {primary_node, allowed_read_regions[], residency_policy, lease_epoch, last_write_ts, preferred_coordinators[]}
• Replica health: node → {region, role, lag_ms, last_heartbeat, is_promotable}
• Store metadata in a strongly consistent metadata service (etcd / Spanner) and expose via streaming watches.

2. Coordinator & cache

• Lightweight coordinators (stateless) colocated with app/gateway; maintain an in-memory TTL cache of tenant routing entries plus subscription to metadata watches for invalidation.
• Cache key: tenant_id. Cached entry includes lease_epoch and version. Cache eviction on version change.
• On cache miss: coordinator fetches metadata from metadata service and populates cache.
• Coordinators attach routing headers (tenant_id, lease_epoch, preferred_region) to RPC to SQL nodes.

Example coordinator lookup (pseudocode):

def route_query(tenant_id, query_type, client_region, session_ctx):
    entry = cache.get(tenant_id)
    if not entry:
        entry = metadata_service.get_tenant(tenant_id)
        cache.set(tenant_id, entry, ttl=5s)
    if query_type == "WRITE":
        return entry.primary_node
    # READ: prefer local replica if allowed and lag<threshold
    local = pick_local_replica(entry.allowed_read_regions, client_region)
    if local and local.lag_ms < entry.max_allowed_lag_ms:
        return local.node
    return entry.primary_node  # fallback for RYW/strong reads

3. Read-your-writes & session guarantees

• Session stickiness: on a session that performed recent writes, coordinator marks session_ctx.min_write_ts = max(min_write_ts, write_ts).
• Replica selection checks replica.lag_ms and replica.latest_applied_ts >= session_ctx.min_write_ts. If none satisfy, route to primary.
• To avoid clock skew, use logical timestamps (hybrid logical clocks or commit sequence numbers) propagated to coordinators.

4. Failover handling

• Promote replica → primary via coordinated controller that:
  • Increments tenant.lease_epoch (atomic update in metadata store).
  • Drains and rejects routing entries with old lease_epoch (coordinator cache invalidation via watch).
  • New primary becomes authoritative; coordinators read new metadata before routing writes.
• During failover, reads allowed only on replicas whose applied_seq >= last_primary_seq or route to new primary to preserve consistency.

5. Scalability & resilience

• Cache: short TTL + metadata watch pushes to minimize stale routing.
• Shard metadata by tenant hash; metadata service must be highly available and consistent.
• Instrumentation: expose per-tenant lag, failed_routes, RYW fallbacks for SLOs.

Trade-offs & notes:

• Serving more reads from local replicas lowers latency but increases RYW fallbacks to primary; mitigate via low-latency replication and optimistic reads that validate min_write_ts.
• Strong metadata consistency simplifies correctness but adds control-plane load; mitigate with caching + watch streaming.
• Use logical timestamps to avoid clock synchronization issues.

This design preserves per-tenant residency, allows fast cached routing, supports safe failover via lease epochs, and enforces read-your-writes by comparing session write timestamps to replica applied positions.

Approach: treat cross-tenant leakage as a data-quality + security incident — combine deterministic checks, statistical anomaly detection, and pragmatic sampling/alerting to surface both clear misroutes (high-confidence) and subtle pattern drift (low-confidence).

Observability signals

• Deterministic: tenant_id → dataset/table mapping violations, FK mismatches, schema-level ownership tags, unexpected tenant_id values (null, global, or other tenant ranges).
• Metrics: per-tenant record counts, row-rate distribution, partition/key cardinality, digest hashes (per-tenant bloom filters / fingerprints), distinct tenant count per topic/stream.
• Logs & traces: CDC event metadata (source LSN, connector id), transformation lineage, commit/offsets.

Sampling & instrumentation

• Inline lightweight checks in CDC connectors to tag events as “suspicious” when tenant_id absent/mismatch; sample full payloads for downstream audit (1% baseline, increase to 100% on anomaly).
• Periodic bulk checks: run nightly scans comparing authoritative tenant mapping (auth DB) to ingested tenant_ids for a rolling window.

Anomaly detection

• Rule-based alerts for deterministic failures (e.g., tenant_id ∉ allowed set) — immediate high-priority.
• Statistical models: per-tenant z-score or EWMA on row-rate and cardinality; change-point detection on distribution of tenant_ids per stream; clustering outliers for new/unexpected tenant_id patterns.
• Thresholding: use adaptive thresholds (percentile-based) rather than fixed counts to handle tenant skew.

Alerting & severity

• High-confidence alerts (deterministic): page on-call, automated pipeline pause/quarantine, require manual ACK.
• Medium/low-confidence alerts (statistical): create tickets, notify data owners via digest, escalate if sustained or correlated with deterministic signals.

Balancing false positives vs negatives

• Prioritize reducing false negatives for high-impact tenants: set stricter rules and lower thresholds for production-critical tenants; for low-risk tenants accept higher FP.
• Use progressive escalation: start with low-noise detection (higher precision) that increases sampling/strictness when signal persists — keeps FP low while catching sustained leaks.
• Maintain feedback loop: annotated incidents feed supervised model or adjust thresholds; run periodic precision/recall reviews.

Playbook & remediation

• Automated isolation: tag downstream storage/queries as “quarantined” when leakage detected; enable rollback or targeted delete if identity verified.
• Auditing: keep immutable audit trail of suspicious events and sampled payloads, with lineage to source LSNs.
• Ownership & runbook: define owners, SLAs, steps (triage, validate tenant mapping, rollback/quarantine, postmortem).

Key trade-offs

• Full payload validation reduces FN but costs throughput and privacy — prefer sampling + targeted full-validation on alerts.
• Adaptive thresholds and tenant-prioritization minimize FP noise while protecting sensitive tenants.

This strategy combines deterministic guards for immediate safety with statistical detection and sampling to catch subtle, emergent leaks while keeping alert fatigue manageable.

Situation: I was asked to lead migration of 5,000 tenants from a legacy monolith to a new multi-tenant data platform that our analytics and product teams depend on.

Plan (high-level):

• Stakeholder communication: I’d create a RACI and communication plan. Weekly steering updates to execs (risk, timeline, budget); bi-weekly working syncs with product, SRE, security, and data consumers; daily standups during cutover windows for ops. Provide a runbook, migration SLA, and a public migration dashboard showing progress and KPIs.
• Risk assessment: Run a risk register with likelihood/impact and mitigations. Top risks: data loss, schema incompatibility, performance regressions, regulatory breaches, and tenant-business impact. Mitigations: automated validations, canary migration, contractual blackout windows, encryption and audit trails, and dedicated rollback paths.
• Migration phases:
  1. Discovery & mapping (2–4 weeks): catalog tenants, data volumes, schemas, owners, SLAs.
  2. Build & test (4–8 weeks): implement pipelines, idempotent ETL, schema translators, integration tests, and synthetic load tests.
  3. Canary (2–3 weeks): migrate 1–2% of tenants (representative set) during low traffic; validate correctness, latency, and consumer workflows.
  4. Gradual rollout (rolling batches): move tenants in cohorts (e.g., 50–200), monitor, pause on issues.
  5. Cutover & decommission: finalize DNS/ingress changes, monitor, then decommission legacy components after 30–90 days of stable operation.
• Rollback strategies:
  • Per-tenant rollback: keep writes dual-piped for a rollback window; use traffic steering (feature flags / routing) to revert a tenant to legacy.
  • Immutable snapshots: take point-in-time backups before each cohort; if data corruption detected, restore to staging and rehydrate.
  • Circuit breakers and automated throttles to stop a cohort progression on thresholds.
• KPIs to track:
  • Data correctness: reconciliation success rate (target 99.99%).
  • Latency: end-to-end pipeline latency percentiles (p50/p95/p99).
  • Throughput: events/sec per tenant and overall.
  • Error rate: ETL failures per million events.
  • Time-to-detect & time-to-repair incidents.
  • Business KPIs: number of impacted users, revenue-affecting incidents.
• Handling resistance from product teams:
  • Empathize and surface concrete concerns; map them to risks and mitigations.
  • Offer guarantees: backward compatibility, SLAs, and a sandbox for product teams to validate.
  • Short feedback loops: invite product reps into canary tests and share data-driven results.
  • If teams still resist, propose phased opt-in with incentives (faster queries, richer metadata) and executive-aligned deadlines.
    Result & learning: This approach minimizes blast radius, provides clear auditability, and uses measurable gates so we only proceed when confidence is proven. It balances speed with safety and creates trust through transparency and data-driven validation.

To reconcile two regionally replicated transaction tables under eventual consistency, use a deterministic, idempotent, and safe approach: stream-scan both tables sorted by a deterministic key, diff records to emit ordered reconciliation operations (upsert or tombstone), write operations to a staging log, validate, then apply idempotently using conditional writes (optimistic concurrency / version checks) and retries.

Approach:

1. Deterministic ordering: sort by (tenant_id, transaction_id, last_modified_timestamp) so both sides produce same canonical sequence.
2. Stream comparison: merge-join the two sorted streams to detect missing, divergent, or duplicate records.
3. Emit operations: for each difference produce an op = {op_type: UPSERT/DELETE, key, payload, version}.
4. Idempotency: include version/hash and apply ops using conditional update: only apply if current_version < op.version (or matches expected hash).
5. Safe apply: write ops to a durable staging table/log, validate sample checksums, then apply in batches with transactions or compare-and-swap; keep audit trail.

Example implementation (simplified):

from typing import Iterator, Dict, List, Tuple
import hashlib
import time

# Placeholder DB read/write primitives:
def stream_table_sorted(table_name: str) -> Iterator[Dict]:
    """
    Yield rows sorted by (tenant_id, tx_id, last_modified).
    In production: use DB index/partition and ORDER BY, or export and sort in streaming fashion.
    """
    raise NotImplementedError

def write_staging_ops(ops: List[Dict]):
    """Append ops to a durable staging table/log."""
    raise NotImplementedError

def conditional_apply(op: Dict) -> bool:
    """
    Apply an op idempotently:
    - For UPSERT: UPDATE ... WHERE key = op.key AND current_version < op.version
      or INSERT if missing with version = op.version
    - For DELETE: mark tombstone only if current_version < op.version
    Returns True if applied or already satisfied, False if conflict (retry).
    """
    raise NotImplementedError

def record_hash(row: Dict) -> str:
    h = hashlib.sha256()
    # deterministic serialization
    for k in sorted(row.keys()):
        h.update(str(k).encode() + b'=' + str(row[k]).encode() + b';')
    return h.hexdigest()


def reconcile(source: str, target: str, batch_size: int = 500):
    src_iter = stream_table_sorted(source)
    tgt_iter = stream_table_sorted(target)

    src = next(src_iter, None)
    tgt = next(tgt_iter, None)

    staging_ops = []

    while src is not None or tgt is not None:
        # pick lower key lexicographically (tenant_id, tx_id)
        def key_of(r):
            return (r['tenant_id'], r['tx_id']) if r else (None, None)

        if tgt is None or (src is not None and key_of(src) < key_of(tgt)):
            # present in source only -> upsert to target
            op = {
                'type': 'UPSERT',
                'key': key_of(src),
                'payload': src,
                'version': src['last_modified'],
                'hash': record_hash(src)
            }
            staging_ops.append(op)
            src = next(src_iter, None)
        elif src is None or (tgt is not None and key_of(tgt) < key_of(src)):
            # present in target only -> maybe delete if source authoritative
            op = {
                'type': 'DELETE',
                'key': key_of(tgt),
                'payload': None,
                'version': tgt['last_modified'],
                'hash': None
            }
            staging_ops.append(op)
            tgt = next(tgt_iter, None)
        else:
            # same key in both: compare hashes or versions
            if record_hash(src) != record_hash(tgt):
                # choose the record with higher last_modified as source-of-truth
                chosen = src if src['last_modified'] >= tgt['last_modified'] else tgt
                op = {
                    'type': 'UPSERT',
                    'key': key_of(src),
                    'payload': chosen,
                    'version': chosen['last_modified'],
                    'hash': record_hash(chosen)
                }
                staging_ops.append(op)
            # else identical -> no-op
            src = next(src_iter, None)
            tgt = next(tgt_iter, None)

        # flush staging ops in deterministic batches
        if len(staging_ops) >= batch_size:
            write_staging_ops(staging_ops)
            apply_batch(staging_ops)
            staging_ops = []

    if staging_ops:
        write_staging_ops(staging_ops)
        apply_batch(staging_ops)


def apply_batch(ops: List[Dict], max_retries: int = 3):
    for op in ops:
        attempt = 0
        while attempt < max_retries:
            ok = conditional_apply(op)
            if ok:
                break
            attempt += 1
            time.sleep(0.1 * (2 ** attempt))
        if attempt == max_retries:
            # escalate: write to dead-letter and alert
            raise RuntimeError(f"Failed to apply op: {op['key']}")

Key points:

• Deterministic ordering ensures repeatable reconciliation runs.
• Staging log provides auditability and safe rollback.
• Conditional/apply-on-version makes operations idempotent and safe under concurrency.
• Use batching, backoff retries, and dead-letter for failures.
• In production, optimize with parallel partitioned scans per tenant range, checksums, and strong metrics (applied count, conflicts).
  Complexity: O(N) streaming over rows; IO-bound. Edge cases: clock skew (use logical clocks/versions), partial failures, tombstone retention, schema drift, large tenants (shard).

Situation: We need to migrate tenants from a single shared-schema database to per-tenant dedicated DBs with minimal downtime.

Strategy overview:

1. Preparation

• Inventory tenants, schema variants, FK/topology, data volume, SLAs.
• Provision target DB instances (templates, config, access).
• Choose CDC tool (Debezium/Kafka, AWS DMS) and schema registry (Avro/JSON Schema).

2. Initial bulk sync

• Run a consistent snapshot per tenant (transactional export or database snapshot) to the target. For large tenants use chunked exports or parallel workers (by PK ranges).
• Load into target with idempotent upserts and tombstone handling.

3. Incremental CDC

• Start CDC on source before cutover; stream changes into a staging Kafka topic and apply to target in order.
• Ensure exactly-once or idempotent writes (use unique change ids, upsert semantics).
• Track source binlog offset per-tenant.

4. Cutover process

• Quiesce writes briefly: route application traffic to a short read-only mode or buffer writes (max window to meet SLA).
• Flush CDC to apply any remaining events; verify offsets are caught up.
• Switch application config / routing to point tenant to dedicated DB.
• Resume writes.

5. Testing & validation

• Continuous checks during CDC: row counts, checksums (per-table per-tenant), sampled record diffs.
• Pre-cutover smoke tests: run critical queries on target and compare results.
• Canary cutover: migrate small set of tenants first and monitor.

6. Rollback plan

• If failure before switching routing: stop CDC, revert application routing, keep sources authoritative.
• If failure after switching: either switch back to source DB (if source still receiving writes via dual-write or buffer) or run reverse sync from target to source.
• Keep backups and point-in-time snapshots for both source and target.

7. Handling schema changes during migration

• Freeze DDL for the migration window if possible.
• If DDL must occur: use backward/forward-compatible migrations:
  • Additive changes first (add columns), make application tolerant to both shapes.
  • Use versioned views or shadow columns; deploy code that writes both schemas (dual-write) until all tenants see new schema.
  • Coordinate DDL with CDC tool—ensure the CDC captures DDL or you apply equivalent DDL on targets in same sequence.
• Run schema validation tooling to detect drift.

Key considerations:

• Idempotency, ordering guarantees, monitoring (lag, error rates), secrets/access, and automation (IaC).
  This approach minimizes downtime, provides clear rollback gates, and keeps data consistent throughout migration.