docs: add BigQuery errors table schema and update README

Add BigQuery schema:
- Create errors table with partitioning and clustering
- Metadata table for tracking loaded files
- Example queries for debugging

Update README:
- Error Store (Dead Letter Queue) section
- Storage structure and usage examples
- BigQuery setup and query examples
- Updated roadmap (error store complete)

Features documented:
- Two error types (validation vs processing)
- 30-day retention with GCS lifecycle
- Full event context for debugging
- BigQuery integration for analysis

Author: prosdev

README.md

@@ -340,6 +340,62 @@ python -m scripts.run_bigquery_loader
 
 See `scripts/bigquery/README.md` and `specs/gcs-bigquery-storage/` for full details.
 
+### Error Store (Dead Letter Queue)
+
+All failed events are stored in a GCS-based dead letter queue for debugging and retry:
+
+**Two Error Types:**
+- **Validation Errors**: Missing required fields, invalid schema
+- **Processing Errors**: Storage failures, unexpected exceptions
+
+**Storage Structure:**
+```
+gs://bucket/errors/
+  date=2026-01-15/
+    error_type=validation/
+      error-20260115-100000-abc123.parquet
+    error_type=processing/
+      error-20260115-100500-def456.parquet
+```
+
+**Create BigQuery Errors Table:**
+```bash
+cd scripts/bigquery
+export PROJECT_ID=my-project DATASET=events
+cat create_errors_table.sql | sed "s/{PROJECT_ID}/$PROJECT_ID/g" | sed "s/{DATASET}/$DATASET/g" | bq query --use_legacy_sql=false
+```
+
+**Query Errors:**
+```sql
+-- Find validation errors in last 24 hours
+SELECT
+  error_message,
+  stream,
+  COUNT(*) as count
+FROM `project.dataset.errors`
+WHERE date >= CURRENT_DATE() - 1
+  AND error_type = 'validation_error'
+GROUP BY error_message, stream
+ORDER BY count DESC;
+
+-- Get processing errors with stack traces
+SELECT
+  timestamp,
+  error_message,
+  JSON_EXTRACT_SCALAR(error_details, '$.exception_type') as exception,
+  JSON_EXTRACT_SCALAR(error_details, '$.stack_trace') as stack_trace
+FROM `project.dataset.errors`
+WHERE error_type = 'processing_error'
+ORDER BY timestamp DESC
+LIMIT 10;
+```
+
+**Key Features:**
+- Never loses events - all failures stored for debugging
+- Automatic 30-day retention (GCS lifecycle rules)
+- Full event context (payload, error, timestamp, stream)
+- Queryable via BigQuery for pattern analysis
+
 ### Custom Storage
 
 Implement the `EventStore` protocol for any backend:
@@ -486,7 +542,7 @@ uv run ruff format src/
 - [x] Prometheus metrics
 - [x] EventSubscriptionCoordinator (dual-path architecture)
 - [x] Hash-based sequencer for consistent ordering
-- [ ] Error handling and dead letter queue (ErrorStore protocol exists, needs implementation)
+- [x] Error store with dead letter queue (GCS-based)
 - [ ] Performance benchmarks (10k+ events/sec)
 
 ### v1.0

pyproject.toml

@@ -49,6 +49,11 @@ dev = [
     "ruff>=0.1.0",
     "mypy>=1.7.0",
     "httpx>=0.25.0",  # For testing FastAPI
+    "locust>=2.20.0",  # Load testing
+    "psutil>=5.9.0",  # Resource monitoring
+    # Type stubs for mypy
+    "types-python-dateutil>=2.8.0",
+    "pandas-stubs>=2.1.0",
 ]
 clickhouse = [
     "clickhouse-driver>=0.2.6",
@@ -79,11 +84,25 @@ strict = true
 warn_return_any = true
 warn_unused_configs = true
 disallow_untyped_defs = true
+# Disable import-untyped for packages without stubs
+disable_error_code = ["import-untyped", "attr-defined"]
+
+[[tool.mypy.overrides]]
+module = "google.cloud"
+ignore_missing_imports = true
 
 [[tool.mypy.overrides]]
 module = "google.cloud.*"
 ignore_missing_imports = true
 
+[[tool.mypy.overrides]]
+module = "prometheus_client.*"
+ignore_missing_imports = true
+
+[[tool.mypy.overrides]]
+module = "pyarrow.*"
+ignore_missing_imports = true
+
 [dependency-groups]
 dev = [
     "types-python-dateutil>=2.9.0.20251115",

scripts/bigquery/create_errors_table.sql

@@ -0,0 +1,74 @@
+-- Create errors table for dead letter queue
+--
+-- Usage:
+--   export PROJECT_ID=your-project
+--   export DATASET=events
+--   cat create_errors_table.sql | sed "s/{PROJECT_ID}/$PROJECT_ID/g" | sed "s/{DATASET}/$DATASET/g" | bq query --use_legacy_sql=false
+
+CREATE TABLE IF NOT EXISTS `{PROJECT_ID}.{DATASET}.errors` (
+  error_id STRING NOT NULL,
+  timestamp TIMESTAMP NOT NULL,
+  error_type STRING NOT NULL,  -- validation_error | processing_error
+  error_message STRING NOT NULL,
+  error_details JSON,
+  stream STRING NOT NULL,
+  original_payload JSON NOT NULL,
+  retry_count INT64 DEFAULT 0,
+  retry_after TIMESTAMP,
+  date DATE NOT NULL
+)
+PARTITION BY date
+CLUSTER BY error_type, stream
+OPTIONS(
+  description="Dead letter queue for failed events. All events that fail validation or processing are stored here with full context for debugging.",
+  partition_expiration_days=30
+);
+
+-- Create metadata table for tracking loaded error files
+CREATE TABLE IF NOT EXISTS `{PROJECT_ID}.{DATASET}._loaded_error_files` (
+  file_path STRING NOT NULL,
+  loaded_at TIMESTAMP NOT NULL DEFAULT CURRENT_TIMESTAMP(),
+  row_count INT64,
+  error_type STRING
+)
+PARTITION BY DATE(loaded_at)
+OPTIONS(
+  description="Metadata tracking which error files have been loaded to prevent duplicates"
+);
+
+-- Example queries for debugging
+
+-- Find validation errors in last 24 hours
+-- SELECT
+--   error_type,
+--   error_message,
+--   stream,
+--   COUNT(*) as count
+-- FROM `{PROJECT_ID}.{DATASET}.errors`
+-- WHERE date >= CURRENT_DATE() - 1
+--   AND error_type = 'validation_error'
+-- GROUP BY error_type, error_message, stream
+-- ORDER BY count DESC;
+
+-- Find processing errors with stack traces
+-- SELECT
+--   timestamp,
+--   error_message,
+--   JSON_EXTRACT_SCALAR(error_details, '$.exception_type') as exception,
+--   JSON_EXTRACT_SCALAR(error_details, '$.stack_trace') as stack_trace,
+--   original_payload
+-- FROM `{PROJECT_ID}.{DATASET}.errors`
+-- WHERE error_type = 'processing_error'
+-- ORDER BY timestamp DESC
+-- LIMIT 10;
+
+-- Error rate by stream
+-- SELECT
+--   stream,
+--   error_type,
+--   COUNT(*) as error_count,
+--   ROUND(COUNT(*) * 100.0 / SUM(COUNT(*)) OVER (PARTITION BY stream), 2) as pct
+-- FROM `{PROJECT_ID}.{DATASET}.errors`
+-- WHERE date >= CURRENT_DATE() - 7
+-- GROUP BY stream, error_type
+-- ORDER BY stream, error_count DESC;

src/eventkit/api/dependencies.py

@@ -11,9 +11,9 @@
 from eventkit.processing.sequencer import HashSequencer
 from eventkit.queues import EventQueue, create_queue
 from eventkit.stores.error_store import ErrorStore
-from eventkit.stores.gcs_error import GCSErrorStore
 from eventkit.stores.event_store import EventStore
 from eventkit.stores.gcs import GCSEventStore
+from eventkit.stores.gcs_error import GCSErrorStore
 
 
 @lru_cache

src/eventkit/loaders/bigquery_loader.py

@@ -8,7 +8,7 @@
 import time
 
 import structlog
-from google.cloud import bigquery, storage  # type: ignore[attr-defined]
+from google.cloud import bigquery, storage
 
 logger = structlog.get_logger(__name__)
 

src/eventkit/processing/processor.py

@@ -14,8 +14,6 @@
 - Lifecycle management (start/stop for graceful shutdown)
 """
 
-from datetime import UTC, datetime
-
 import structlog
 
 from eventkit.adapters.base import EventAdapter

src/eventkit/queues/pubsub.py

@@ -19,7 +19,7 @@
 import logging
 from typing import TYPE_CHECKING
 
-from google.cloud import pubsub_v1  # type: ignore[attr-defined]
+from google.cloud import pubsub_v1
 from google.cloud.pubsub_v1.subscriber.message import Message
 
 from eventkit.config import Settings

src/eventkit/stores/error_store.py

@@ -1,6 +1,5 @@
 """Error storage protocol and implementations."""
 
-from datetime import datetime
 from typing import Any, Literal, Protocol
 
 import structlog

src/eventkit/stores/gcs.py

@@ -12,10 +12,10 @@
 from uuid import uuid4
 
 import pandas as pd
-import pyarrow as pa  # type: ignore[import-untyped]
-import pyarrow.parquet as pq  # type: ignore[import-untyped]
+import pyarrow as pa
+import pyarrow.parquet as pq
 import structlog
-from google.cloud import storage  # type: ignore[attr-defined]
+from google.cloud import storage
 from tenacity import retry, stop_after_attempt, wait_exponential
 
 from eventkit.schema.events import IdentifyEvent, PageEvent, TrackEvent, TypedEvent

src/eventkit/stores/gcs_error.py

@@ -4,8 +4,8 @@
 from typing import Any, Literal
 
 import pandas as pd
-import pyarrow as pa  # type: ignore[import-not-found]
-import pyarrow.parquet as pq  # type: ignore[import-not-found]
+import pyarrow as pa
+import pyarrow.parquet as pq
 from google.cloud import storage
 from tenacity import retry, stop_after_attempt, wait_exponential