Streaming client improvements and snowflake loader features

[fordN]

I apologize ahead of time for the size of this PR 😬 . I'm now adding a detailed description here with tips for reviewing to hopefully make it more digestible.

Overview

This PR adds streaming infrastructure features to amp-python for comprehensive state management, resilience features, and a significantly enhanced Snowflake loader. The changes enable resumable, fault-tolerant streaming with automatic gap detection, reorg handling, and label enrichment.

---

Review Strategy

I recommend reviewing in this order:

1. Core Concepts (20 min)

Start here to understand the foundation:

• src/amp/streaming/state.py - Unified state management
• src/amp/streaming/types.py - Core types: BlockRange, ResponseBatch, BatchIdentifier
• docs/resilience.md - Architecture overview

2. Key Features (30 min)

Review the main capabilities:

• src/amp/loaders/base.py (lines 200-400) - Streaming support in base loader
• src/amp/streaming/parallel.py (lines 350-450) - Gap detection & resume optimization
• src/amp/config/label_manager.py - CSV label enrichment
• src/amp/streaming/resilience.py - Retry, backpressure, rate limiting

3. Snowflake Implementation (20 min)

The largest single file change:

• src/amp/loaders/implementations/snowflake_loader.py - Persistent state, COPY INTO, parallel loading

4. Tests & Validation (15 min)

Verify comprehensive coverage:

• tests/unit/test_stream_state.py - State management tests
• tests/integration/test_checkpoint_resume.py - End-to-end resume scenarios
• tests/integration/test_resilient_streaming.py - Resilience with real databases

5. Other stuffs (10 min)

• Apps, docs, Docker configs, benchmarks

---

Key Features

1. Unified Stream State Management

Why: Simplifies resumability + idempotency into one system
Files: src/amp/streaming/state.py, sql/snowflake_stream_state.sql

• StreamStateStore interface with in-memory, null, and DB-backed implementations
• BatchIdentifier with hash-based uniqueness for reorg detection
• Automatic gap detection and intelligent backfill
• Support for multiple networks and tables

Example:

state_store = InMemoryStreamStateStore()
state_store.mark_processed(conn, table, batch_ids)
resume_pos = state_store.get_resume_position(conn, table, detect_gaps=True)

2. Label Enrichment System

Why: Join blockchain data with metadata (token info, labels, etc.)
Files: src/amp/config/label_manager.py, docs/label_manager.md

• Load CSV labels with automatic hex→binary conversion
• Efficient PyArrow-based joins during streaming
• Thread-safe label storage and retrieval

Example:

client.add_label('tokens', 'eth_mainnet_token_metadata.csv')
# Automatically joins on matching address columns

3. Resilience Features

Why: Production workloads need fault tolerance
Files: src/amp/streaming/resilience.py, docs/resilience.md

• Exponential backoff with jitter (configurable max retries)
• Adaptive rate limiting (responds to 429s and timeouts)
• Backpressure detection and circuit breakers
• Transient vs permanent error classification

4. Parallel Execution Enhancements

Why: Fill historical gaps while maintaining streaming
Files: src/amp/streaming/parallel.py

• Intelligent gap detection and prioritization
• Resume optimization (adjusts min_block based on progress)
• Gap-aware partitioning strategies
• Hybrid mode: parallel backfill + streaming updates

Example:

config = ParallelConfig(
    table_name='erc20_transfers',
    min_block=10_000_000,
    max_block=21_000_000,
    num_workers=4
)
executor.load_with_resume_optimization(config, detect_gaps=True)

5. Enhanced Snowflake Loader

Why: Enterprise-grade streaming to Snowflake
Files: src/amp/loaders/implementations/snowflake_loader.py

• Persistent state via Snowflake tables (survives restarts)
• Parquet COPY INTO for high-performance bulk loading (via stages)
• Parallel workers with proper connection pooling
• Reorg handling options delete reorged data (legacy behavior) or keep history and mark as reorged
• Reorg handling with batch-level invalidation

---

Testing summary

Coverage

• 172 unit tests - Pure logic, minimal mocking
• 133 integration tests - Real database interactions via testcontainers
• All loaders updated: PostgreSQL, Redis, Snowflake, Delta, Iceberg, LMDB

Key Test Files

• tests/unit/test_stream_state.py - State management
• tests/unit/test_resilience.py - Retry, backpressure, rate limiting
• tests/integration/test_checkpoint_resume.py - Resume scenarios
• tests/integration/test_resilient_streaming.py - Fault injection

Run Tests

# Install all dependencies (if haven't yet)
uv build
uv sync --all-groups

# Unit tests (fast)
make test-unit

# Integration tests (require Docker or similar such as Colima)
make test-integration

# All tests
make test-all

---

Architecture Decisions

Why Unified State Management?

Previous design (replaced by this) had separate systems for checkpoints (resume) and processed ranges (idempotency). This created:

• Duplication (two systems tracking similar data)
• Complexity (had to manage both)
• Inconsistency (could get out of sync)

Solution: StreamStateStore provides both capabilities with a single source of truth.

Why Hash-based Batch IDs?

Block ranges alone aren't unique during reorgs (same range, different chain state). Hash-based IDs:

• Include block hashes for uniqueness across reorgs
• Enable precise invalidation (delete only affected batches)
• Compact (16-char hex = 64 bits)

---

New Files

Core Infrastructure

• src/amp/streaming/state.py - Unified state management
• src/amp/streaming/resilience.py - Retry, backpressure, rate limiting
• src/amp/config/label_manager.py - CSV label enrichment
• sql/snowflake_stream_state.sql - Snowflake state schema

Applications

• apps/snowflake_parallel_loader.py - Production-ready parallel loader
• apps/test_erc20_labeled_parallel.py - Example with label enrichment
• apps/queries/erc20_transfers.sql - Sample query

Documentation

• docs/resilience.md - Complete resilience architecture guide
• docs/label_manager.md - Label enrichment guide
• apps/SNOWFLAKE_LOADER_GUIDE.md - Snowflake loader tutorial
• apps/queries/README.md - Query examples

Infrastructure

• Dockerfile, Dockerfile.snowflake - Containerized deployment
• k8s/deployment.yaml - Kubernetes deployment
• .github/workflows/docker-publish.yml - Docker CI/CD

---

Key Commit Groups

Foundation (Commits 1-3)

• Label management system
• Unified state management
• Resilience features

Core Streaming (Commits 4-6)

• Base loader improvements
• Parallel execution enhancements
• Client integration

Loader Updates (Commit 7)

• All loaders updated for new interface
• Adds _amp_batch_id metadata column
• Reorg handling support

Snowflake (Commits 8-9)

• Enhanced Snowflake loader with persistent state
• Parallel loading applications

Testing (Commits 10-11)

• 172 unit tests
• 133 integration tests

Polish (Commits 12-18)

• Documentation
• Docker/K8s configs
• Performance benchmarks
• Linting fixes
• Bug fix: Redis reorg with string data structure

---

Performance

See performance_benchmarks.json for detailed metrics.

Highlights:

• Snowflake COPY INTO: 10-100x faster than INSERT
• Parallel loading: Linear speedup with workers (tested up to 8 workers)
• Label joins: ~19ms overhead for 3K rows (negligible)
• Resume optimization: Automatic gap detection (no manual intervention)

---