Graceful Shutdown and State Persistence: Building Resilient WebSocket Data Pipelines That Never Lose Data

---

"The server restarted at 3:47 AM. Your WebSocket dropped. You didn't notice until morning."

That sentence has ended more than a few quantitative trading careers—or at least a few weekends. In live trading systems, a WebSocket disconnection is not just an inconvenience. It is a data loss event. The moment your connection drops, you are flying blind: no order book updates, no trade ticks, no depth snapshots. If your system has not persisted its state, you are starting from zero.

This article is a practitioner's guide to building WebSocket pipelines that survive restarts gracefully. We cover signal handling, checkpoint persistence with SQLite, and a resumable transmission pattern that bridges the gap between disconnection and reconnection. The code is production-grade and designed for real-time market data ingestion pipelines—exactly the kind of environment where 30 seconds of data loss costs money.

---

Why WebSocket Data Loss Happens

Before solving the problem, it helps to understand the failure taxonomy. WebSocket disconnections fall into three categories:

1. Unexpected server-side termination. The data provider closes the connection without notice—due to server maintenance, a deployed update, or a load-shedding event. The client receives no close frame.

2. Network path disruption. The client machine loses internet connectivity—transient wifi dropout, a VPN tunnel failure, or a cloud instance eviction. The TCP connection hangs until the OS enforces a keepalive timeout, which can take 30–120 seconds.

3. Graceful shutdown of the client process. Someone runs kill -9, deploys a new container version, or restarts the machine for an OS patch. The process terminates immediately with no opportunity to flush state.

Most developers implement reconnection logic for case 1 and case 2. Case 3—the voluntary process shutdown—is where most systems fail catastrophically, because the shutdown handler never fires.

The solution is not a better reconnect strategy. The solution is a checkpoint-before-shutdown pattern: before the process exits for any reason, persist the last known state to durable storage. On startup, read the checkpoint and resume from where you left off.

---

Architecture Overview

The system consists of four interacting components:

┌─────────────────────────────────────────────────────────────┐
│                    Application Process                       │
│  ┌──────────┐  ┌──────────────┐  ┌────────────────────┐    │
│  │  Signal  │──│  Shutdown    │──│  Checkpoint        │    │
│  │  Handler │  │  Coordinator │  │  Manager           │    │
│  └──────────┘  └──────────────┘  └─────────┬──────────┘    │
│                                            │                 │
│  ┌──────────────┐  ┌──────────────┐        │                 │
│  │  WebSocket   │──│  Data        │─────────┘                 │
│  │  Client      │  │  Processor   │                           │
│  └──────────────┘  └──────────────┘                           │
└─────────────────────────────────────────────────────────────┘
                         │
                         ▼
              ┌──────────────────────┐
              │   SQLite DB         │
              │   (checkpoint.db)   │
              │   - last_seq        │
              │   - last_timestamp  │
              │   - buffer_payload  │
              └──────────────────────┘

The Shutdown Coordinator receives termination signals and coordinates an orderly shutdown sequence: stop accepting new data, flush the checkpoint, then exit. The Checkpoint Manager writes the current sequence number, timestamp, and a configurable payload buffer to SQLite. On startup, the application reads the checkpoint and issues a resumable query to the data provider.

---

Signal Handling: Catching the Shutdown Signal

Unix-like systems deliver three signals relevant to graceful shutdown:

Signal	Default behavior	When sent
SIGTERM	Process terminates	kill, Kubernetes pod eviction, Docker stop
SIGINT	Process terminates	Ctrl+C in terminal
SIGSEGV	Core dump + terminate	Segmentation fault (unrecoverable)

You can only handle SIGTERM and SIGINT gracefully. SIGSEGV means your process is already corrupted.

import signal
import sys
import threading
from typing import Callable, Optional


class ShutdownCoordinator:
    """Coordinates graceful shutdown across all system components."""

    def __init__(self):
        self._shutdown_requested = threading.Event()
        self._shutdown_complete = threading.Event()
        self._handlers: list[Callable[[], None]] = []
        self._registered = False

    def request_shutdown(self) -> None:
        """Called by signal handler or internal shutdown trigger."""
        if self._shutdown_requested.is_set():
            return  # Prevent double-trigger
        self._shutdown_requested.set()
        self._execute_cleanup()
        self._shutdown_complete.set()

    def _execute_cleanup(self) -> None:
        """Run all registered cleanup handlers in reverse order of registration."""
        for handler in reversed(self._handlers):
            try:
                handler()
            except Exception as e:
                # Log but do not re-raise — cleanup should never block shutdown
                print(f"Cleanup handler error: {e}", file=sys.stderr)

    def register_handler(self, handler: Callable[[], None]) -> None:
        """Register a cleanup handler. Called in LIFO order during shutdown."""
        self._handlers.append(handler)

    @property
    def is_shutting_down(self) -> bool:
        return self._shutdown_requested.is_set()

    def wait_for_shutdown(self, timeout: Optional[float] = None) -> bool:
        """Block until shutdown completes or timeout expires."""
        return self._shutdown_complete.wait(timeout=timeout)


# Global singleton
_coordinator = ShutdownCoordinator()


def _handle_signal(signum: int, frame) -> None:
    """Unix signal handler — must not raise exceptions."""
    sig_name = signal.Signals(signum).name
    print(f"Received {sig_name}, initiating graceful shutdown...")
    _coordinator.request_shutdown()


def register_signal_handlers() -> None:
    """Install signal handlers for SIGTERM and SIGINT."""
    if _coordinator._registered:
        return
    signal.signal(signal.SIGTERM, _handle_signal)
    signal.signal(signal.SIGINT, _handle_signal)
    _coordinator._registered = True

⚠️ Engineering note: Signal handlers have strict constraints in Python. They must not raise exceptions and should return as quickly as possible. The actual cleanup work is deferred to _execute_cleanup, which runs in the main thread via _coordinator.request_shutdown(). If your process is stuck in an infinite loop or a blocking I/O call when the signal arrives, the signal handler will execute but cleanup will not run until the blocking call releases. Always set timeouts on blocking operations.

---

Checkpoint Persistence with SQLite

SQLite is the right tool for this job. It is ACID-compliant, requires no server, handles concurrent readers perfectly, and has sub-millisecond write latency on local SSDs. For a checkpoint writer that flushes on shutdown (at most a few times per day), SQLite is dramatically simpler than setting up PostgreSQL or Redis.

Schema Design

import sqlite3
import json
import time
from dataclasses import dataclass
from pathlib import Path
from typing import Any, Optional


@dataclass
class Checkpoint:
    """Represents the persisted state of the data pipeline."""
    last_sequence: int          # Last processed sequence number
    last_timestamp: float       # Unix timestamp of last update
    buffer_payload: str         # JSON-serialized recent data (for gap fill)
    channel_state: str          # JSON-serialized per-channel metadata


class CheckpointManager:
    """Persists and retrieves pipeline checkpoint state using SQLite."""

    def __init__(self, db_path: Path):
        self.db_path = db_path
        self._init_db()

    def _init_db(self) -> None:
        """Create the checkpoints table if it does not exist."""
        with sqlite3.connect(self.db_path) as conn:
            conn.execute("""
                CREATE TABLE IF NOT EXISTS checkpoints (
                    id              INTEGER PRIMARY KEY CHECK (id = 1),
                    last_sequence   INTEGER NOT NULL DEFAULT 0,
                    last_timestamp  REAL NOT NULL,
                    buffer_payload  TEXT NOT NULL DEFAULT '[]',
                    channel_state   TEXT NOT NULL DEFAULT '{}',
                    updated_at      REAL NOT NULL
                )
            """)
            # Ensure exactly one row exists
            conn.execute("""
                INSERT OR IGNORE INTO checkpoints (id, last_sequence, last_timestamp, updated_at)
                VALUES (1, 0, 0.0, 0.0)
            """)
            conn.commit()

    def write(self, checkpoint: Checkpoint) -> None:
        """Persist the current checkpoint to SQLite."""
        with sqlite3.connect(self.db_path) as conn:
            conn.execute("""
                UPDATE checkpoints SET
                    last_sequence  = ?,
                    last_timestamp = ?,
                    buffer_payload = ?,
                    channel_state  = ?,
                    updated_at     = ?
                WHERE id = 1
            """, (
                checkpoint.last_sequence,
                checkpoint.last_timestamp,
                checkpoint.buffer_payload,
                checkpoint.channel_state,
                time.time(),
            ))
            conn.commit()

    def read(self) -> Optional[Checkpoint]:
        """Read the most recent checkpoint, or None if no checkpoint exists."""
        with sqlite3.connect(self.db_path) as conn:
            conn.row_factory = sqlite3.Row
            row = conn.execute("SELECT * FROM checkpoints WHERE id = 1").fetchone()
            if row is None or row["last_sequence"] == 0:
                return None
            return Checkpoint(
                last_sequence=row["last_sequence"],
                last_timestamp=row["last_timestamp"],
                buffer_payload=row["buffer_payload"],
                channel_state=row["channel_state"],
            )

    def clear(self) -> None:
        """Reset the checkpoint after successful resynchronization."""
        self.write(Checkpoint(
            last_sequence=0,
            last_timestamp=0.0,
            buffer_payload="[]",
            channel_state="{}",
        ))

⚠️ Engineering note: The buffer_payload field stores a JSON-serialized list of the most recent N messages. This serves as a short-term buffer for gap filling—if you disconnect for 5 seconds and the provider supports sequence-based gap fill, you can request exactly the missing messages rather than replaying a large time range. Size this buffer to cover your expected worst-case reconnection time (typically 30–120 seconds at most). A 10 KB buffer at 100 bytes per message holds ~100 messages, which is sufficient for most market data use cases.

---

Integrating Checkpoint Management into the Data Pipeline

With signal handling and checkpoint persistence in place, the next step is integrating them into the WebSocket client lifecycle.

The WebSocket Client with State Tracking

import json
import os
import time
import random
import threading
import websocket  # websocket-client library
from dataclasses import dataclass, field, asdict
from pathlib import Path
from typing import Callable, Optional, Any
from enum import Enum

from .signal_handler import _coordinator, register_signal_handlers
from .checkpoint import CheckpointManager, Checkpoint


class ConnectionState(Enum):
    DISCONNECTED = "disconnected"
    CONNECTING = "connecting"
    CONNECTED = "connected"
    RECONNECTING = "reconnecting"
    SHUTTING_DOWN = "shutting_down"


@dataclass
class MarketDataMessage:
    """Represents a single market data update."""
    sequence: int
    timestamp: float
    channel: str
    symbol: str
    data: dict[str, Any]


@dataclass
class WebSocketDataClient:
    """WebSocket client with graceful shutdown and checkpoint recovery."""

    api_key: str
    base_url: str = "wss://api.tickdb.ai/ws/market"
    checkpoint_manager: Optional[CheckpointManager] = None
    buffer_size: int = 100
    max_reconnect_attempts: int = 10
    base_reconnect_delay: float = 1.0
    max_reconnect_delay: float = 60.0

    # Internal state
    _state: ConnectionState = field(default=ConnectionState.DISCONNECTED)
    _ws: Optional[websocket.WebSocketApp] = field(default=None, repr=False)
    _thread: Optional[threading.Thread] = field(default=None, repr=False)
    _message_buffer: list[MarketDataMessage] = field(default_factory=list)
    _last_sequence: int = 0
    _last_timestamp: float = 0.0
    _channel_metadata: dict[str, Any] = field(default_factory=dict)
    _reconnect_attempts: int = 0
    _lock: threading.Lock = field(default_factory=threading.Lock)

    def __post_init__(self) -> None:
        register_signal_handlers()

        # Register the checkpoint flush handler with the shutdown coordinator
        _coordinator.register_handler(self._flush_checkpoint)

    def connect(self, subscriptions: list[str]) -> None:
        """Connect to the WebSocket endpoint and subscribe to channels."""
        if self._state in (ConnectionState.CONNECTED, ConnectionState.CONNECTING):
            return

        # Load checkpoint if available
        checkpoint = self._load_checkpoint()
        if checkpoint:
            self._last_sequence = checkpoint.last_sequence
            self._last_timestamp = checkpoint.last_timestamp
            print(f"Resuming from checkpoint: seq={self._last_sequence}, ts={self._last_timestamp}")

        self._state = ConnectionState.CONNECTING
        params = f"?api_key={self.api_key}"
        if checkpoint and checkpoint.last_sequence > 0:
            params += f"&since_seq={checkpoint.last_sequence}"

        url = f"{self.base_url}{params}"
        headers = {"X-API-Key": self.api_key}

        self._ws = websocket.WebSocketApp(
            url,
            header=headers,
            on_open=self._on_open,
            on_message=self._on_message,
            on_error=self._on_error,
            on_close=self._on_close,
        )

        self._thread = threading.Thread(target=self._ws.run_forever, daemon=True)
        self._thread.start()

    def _on_open(self, ws: websocket.WebSocketApp) -> None:
        """Called when WebSocket connection is established."""
        with self._lock:
            self._state = ConnectionState.CONNECTED
            self._reconnect_attempts = 0
        print("WebSocket connected")

    def _on_message(self, ws: websocket.WebSocketApp, raw_message: str) -> None:
        """Process incoming WebSocket messages."""
        if _coordinator.is_shutting_down:
            return

        try:
            msg = json.loads(raw_message)
        except json.JSONDecodeError:
            return

        # Handle ping from server ( heartbeat mechanism )
        if msg.get("type") == "ping":
            ws.send(json.dumps({"type": "pong", "timestamp": time.time()}))
            return

        # Process market data message
        if "data" in msg:
            for item in msg["data"]:
                sequence = item.get("seq", 0)
                timestamp = item.get("t", time.time())
                channel = item.get("ch", "")
                symbol = item.get("sym", "")
                data = item.get("d", {})

                message = MarketDataMessage(
                    sequence=sequence,
                    timestamp=timestamp,
                    channel=channel,
                    symbol=symbol,
                    data=data,
                )

                with self._lock:
                    self._last_sequence = sequence
                    self._last_timestamp = timestamp

                    # Maintain circular buffer of recent messages
                    self._message_buffer.append(message)
                    if len(self._message_buffer) > self.buffer_size:
                        self._message_buffer = self._message_buffer[-self.buffer_size:]

                # Dispatch to application handler
                self._dispatch_message(message)

    def _dispatch_message(self, message: MarketDataMessage) -> None:
        """Override this method to handle incoming messages."""
        # Default implementation logs the message
        print(f"[{message.channel}] {message.symbol}: seq={message.sequence}, ts={message.timestamp}")

    def _on_error(self, ws: websocket.WebSocketApp, error: Exception) -> None:
        """Handle WebSocket errors."""
        print(f"WebSocket error: {error}")
        if _coordinator.is_shutting_down:
            return
        self._schedule_reconnect()

    def _on_close(self, ws: websocket.WebSocketApp, close_status_code: int, close_msg: str) -> None:
        """Handle WebSocket connection closure."""
        with self._lock:
            self._state = ConnectionState.DISCONNECTED
        print(f"WebSocket closed: code={close_status_code}, msg={close_msg}")

        if not _coordinator.is_shutting_down:
            self._schedule_reconnect()

    def _schedule_reconnect(self) -> None:
        """Schedule a reconnection attempt with exponential backoff and jitter."""
        with self._lock:
            if self._state == ConnectionState.RECONNECTING:
                return
            if self._reconnect_attempts >= self.max_reconnect_attempts:
                print("Max reconnection attempts reached. Giving up.")
                return
            self._state = ConnectionState.RECONNECTING
            self._reconnect_attempts += 1

        # Exponential backoff with jitter
        delay = min(self.base_reconnect_delay * (2 ** (self._reconnect_attempts - 1)), self.max_reconnect_delay)
        jitter = random.uniform(0, delay * 0.1)
        reconnect_delay = delay + jitter

        print(f"Reconnecting in {reconnect_delay:.2f}s (attempt {self._reconnect_attempts}/{self.max_reconnect_attempts})")

        def delayed_reconnect():
            time.sleep(reconnect_delay)
            if not _coordinator.is_shutting_down:
                # Re-subscribe to the same channels
                self._state = ConnectionState.DISCONNECTED
                # Note: in a real implementation, track subscribed channels and re-subscribe

        threading.Thread(target=delayed_reconnect, daemon=True).start()

    def _load_checkpoint(self) -> Optional[Checkpoint]:
        """Load checkpoint from SQLite on startup."""
        if self.checkpoint_manager is None:
            return None
        return self.checkpoint_manager.read()

    def _flush_checkpoint(self) -> None:
        """Flush current state to checkpoint — called during graceful shutdown."""
        if self.checkpoint_manager is None:
            return

        buffer_json = json.dumps([asdict(m) for m in self._message_buffer])
        channel_json = json.dumps(self._channel_metadata)

        checkpoint = Checkpoint(
            last_sequence=self._last_sequence,
            last_timestamp=self._last_timestamp,
            buffer_payload=buffer_json,
            channel_state=channel_json,
        )

        self.checkpoint_manager.write(checkpoint)
        print(f"Checkpoint flushed: seq={checkpoint.last_sequence}, ts={checkpoint.last_timestamp}")

        with self._lock:
            self._state = ConnectionState.SHUTTING_DOWN

        if self._ws:
            self._ws.close()
            print("WebSocket connection closed gracefully")

    def disconnect(self) -> None:
        """Initiate graceful disconnect."""
        _coordinator.request_shutdown()

⚠️ Engineering note: The _dispatch_message method is a stub. In a production system, this is where you would implement your data processing logic—writing to a database, forwarding to a message queue, updating in-memory state, or triggering trading signals. Keep this method fast and non-blocking. Any blocking I/O here will delay message processing and can cause the WebSocket receive buffer to overflow, triggering unwanted disconnections.

---

Resumable Data Retrieval: Bridging the Gap

Checkpoint persistence solves the problem of saving state on shutdown. But the other half of the problem is: after reconnecting, how do you fill the gap between the last known sequence and the current live feed?

This requires two things: a sequence-aware data provider and a gap-fill query on reconnect.

Querying Historical Data for Gap Fill

Most professional market data APIs—including TickDB—provide a way to retrieve historical data by sequence number or timestamp. The pattern is:

1. On reconnect, read the checkpoint to get last_sequence.
2. Issue a gap-fill query: GET /v1/market/trades?since_seq={last_sequence + 1}
3. Apply the gap-fill data to your state before switching to live WebSocket feed.

import os
import requests

# Load from environment — never hardcode API keys
TICKDB_API_KEY = os.environ.get("TICKDB_API_KEY")
if not TICKDB_API_KEY:
    raise ValueError("TICKDB_API_KEY environment variable is not set")


def gap_fill_trades(symbol: str, since_seq: int, limit: int = 1000) -> list[dict]:
    """
    Retrieve trades since the given sequence number for gap filling.

    This endpoint returns historical trades that may have been missed
    during a disconnection window.
    """
    url = "https://api.tickdb.ai/v1/market/trades"
    headers = {"X-API-Key": TICKDB_API_KEY}

    params = {
        "symbol": symbol,
        "since_seq": since_seq + 1,  # Resume from the next sequence after checkpoint
        "limit": limit,
    }

    try:
        response = requests.get(url, headers=headers, params=params, timeout=(3.05, 10))
        response.raise_for_status()
        data = response.json()

        if data.get("code") == 0:
            return data.get("data", [])
        else:
            # Handle API error codes per TickDB error reference
            raise RuntimeError(f"API error {data.get('code')}: {data.get('message')}")

    except requests.exceptions.Timeout:
        raise RuntimeError(f"Request timed out while fetching gap fill data for {symbol}")
    except requests.exceptions.HTTPError as e:
        if e.response.status_code == 429:
            retry_after = int(e.response.headers.get("Retry-After", 5))
            raise RuntimeError(f"Rate limited. Retry after {retry_after} seconds.")
        raise RuntimeError(f"HTTP error {e.response.status_code} fetching gap fill data")


def fetch_kline_for_backfill(symbol: str, start_time: float, end_time: float, interval: str = "1m") -> list[dict]:
    """
    Retrieve historical OHLCV candles for backfill after reconnection.

    Use this when the provider does not support sequence-based gap fill,
    and you need to rebuild state from time-based queries.
    """
    url = "https://api.tickdb.ai/v1/market/kline"
    headers = {"X-API-Key": TICKDB_API_KEY}

    params = {
        "symbol": symbol,
        "interval": interval,
        "start_time": int(start_time * 1000),  # API expects milliseconds
        "end_time": int(end_time * 1000),
        "limit": 1000,
    }

    response = requests.get(url, headers=headers, params=params, timeout=(3.05, 10))
    response.raise_for_status()
    data = response.json()

    if data.get("code") == 0:
        return data.get("data", [])
    else:
        raise RuntimeError(f"API error {data.get('code')}: {data.get('message')}")

⚠️ Engineering note: The timeout=(3.05, 10) parameter on requests.get uses the tuple form: (connect_timeout, read_timeout). The 3.05-second connect timeout is slightly above the 3-second mark to avoid race conditions with server-side rate limit windows. The 10-second read timeout prevents the process from hanging on slow responses. Adjust these based on your network conditions and SLA requirements.

---

Application Startup: The Full Recovery Sequence

Putting it all together, here is the complete startup sequence that handles graceful recovery:

from pathlib import Path


def main():
    """Application entry point with full checkpoint recovery."""
    db_path = Path.home() / ".tickdb_pipeline" / "checkpoint.db"
    db_path.parent.mkdir(parents=True, exist_ok=True)

    checkpoint_manager = CheckpointManager(db_path)
    checkpoint = checkpoint_manager.read()

    if checkpoint:
        print(f"Found checkpoint: resuming from sequence {checkpoint.last_sequence}")
        # Restore message buffer for potential reprocessing
        restored_messages = [
            MarketDataMessage(**m) for m in json.loads(checkpoint.buffer_payload)
        ]
        print(f"Restored {len(restored_messages)} buffered messages")

        # Fetch gap fill from provider
        try:
            gap_data = gap_fill_trades("BTC-USD", checkpoint.last_sequence)
            print(f"Retrieved {len(gap_data)} gap-fill records")
            for record in gap_data:
                process_trade_record(record)
        except Exception as e:
            print(f"Gap fill failed: {e}")
            print("Proceeding with live feed — some data may be missing")

    # Initialize and connect the WebSocket client
    client = WebSocketDataClient(
        api_key=TICKDB_API_KEY,
        checkpoint_manager=checkpoint_manager,
    )

    # Connect to desired channels
    client.connect(subscriptions=["trades.BTC-USD", "depth.BTC-USD"])

    # Block the main thread — the coordinator handles graceful shutdown
    _coordinator.wait_for_shutdown()
    print("Shutdown complete")


def process_trade_record(record: dict) -> None:
    """Process a single trade record from gap fill or live feed."""
    # Placeholder — replace with actual business logic
    print(f"Processing trade: {record.get('sym')} @ {record.get('p')}")


if __name__ == "__main__":
    main()

---

Testing the Graceful Shutdown Path

A graceful shutdown implementation is only as good as its test coverage. You need to verify that:

1. SIGTERM triggers checkpoint flush. Send kill -TERM to the process and verify the SQLite checkpoint was written.
2. SIGINT triggers checkpoint flush. Send kill -INT and verify the same.
3. No race conditions under load. Flush the checkpoint while the WebSocket is receiving a high-frequency stream. Verify no data is lost or corrupted.
4. Restart recovers correctly. Kill the process, restart it, and verify the gap-fill query retrieves exactly the missing sequence range.

import subprocess
import time
import signal


def test_graceful_shutdown():
    """Integration test: verify SIGTERM triggers checkpoint flush."""
    # Start the pipeline as a subprocess
    proc = subprocess.Popen(
        ["python", "-m", "my_pipeline.main"],
        env={**os.environ, "TICKDB_API_KEY": os.environ["TICKDB_API_KEY"]},
        stdout=subprocess.PIPE,
        stderr=subprocess.PIPE,
    )

    # Let it run and accumulate data
    time.sleep(5)

    # Send SIGTERM
    proc.send_signal(signal.SIGTERM)
    proc.wait(timeout=10)

    # Verify checkpoint was written
    checkpoint_manager = CheckpointManager(Path.home() / ".tickdb_pipeline" / "checkpoint.db")
    checkpoint = checkpoint_manager.read()

    assert checkpoint is not None, "Checkpoint was not written"
    assert checkpoint.last_sequence > 0, "Checkpoint sequence is zero — data may not have been received"
    print(f"Test passed: checkpoint written with seq={checkpoint.last_sequence}")


def test_restart_recovery():
    """Integration test: verify restart recovers from checkpoint."""
    # Run the pipeline once and let it write a checkpoint
    # (Same setup as test_graceful_shutdown)

    # Run again and verify it reports resuming from the checkpoint
    proc = subprocess.Popen(
        ["python", "-m", "my_pipeline.main"],
        env={**os.environ, "TICKDB_API_KEY": os.environ["TICKDB_API_KEY"]},
        stdout=subprocess.PIPE,
        stderr=subprocess.PIPE,
        text=True,
    )

    stdout, _ = proc.communicate(timeout=10)

    assert "Resuming from checkpoint" in stdout, "Restart did not recover from checkpoint"
    print("Test passed: restart recovery working correctly")

⚠️ Engineering note: SIGKILL (kill -9) cannot be caught by any process. The shutdown coordinator will never run, and no checkpoint will be written. This is by design in Unix. If you need to survive SIGKILL, you need an external watchdog process that monitors the pipeline's heartbeat and writes the checkpoint independently. For most production deployments, however, SIGKILL during normal operations indicates a deployment or orchestration issue that should be addressed at the infrastructure level, not patched in application code.

---

Deployment Recommendations by Scale

Scale	Checkpoint storage	Recommended approach
Individual developer	SQLite in ~/.pipeline/	Single-process pipeline, checkpoint flush on every SIGTERM
Team / shared server	SQLite on network-attached storage (NAS)	Shared checkpoint file with file locking; consider Redis for higher throughput
Institutional / multi-instance	PostgreSQL or TimescaleDB	Centralized checkpoint store; use optimistic locking to prevent race conditions across instances

For single-instance pipelines running on cloud VMs or containers, SQLite is the correct default choice. For Kubernetes deployments where pods can be preempted at any time, ensure your checkpoint flush handler runs in under 5 seconds—Kubernetes sends SIGTERM and then waits for the terminationGracePeriodSeconds (default 30 seconds) before sending SIGKILL. Design your flush to complete within 10 seconds to leave margin.

---

What Can Go Wrong: Failure Mode Inventory

Failure mode	Symptom	Root cause	Mitigation
No checkpoint written	Restart always starts from sequence 0	Signal handler not registered; kill -9 used	Verify handler registration on startup; monitor updated_at in checkpoint table
Stale checkpoint	Pipeline resumes from old sequence, duplicates processed	Checkpoint written but gap-fill query not issued	Verify startup code calls gap_fill_trades; deduplicate by sequence number
Buffer overflow	Old messages dropped before shutdown	Buffer size too small for shutdown window	Size buffer to cover 2× expected worst-case reconnection time
Race: message received after flush	Some messages lost between last flush and socket close	WebSocket receive thread and shutdown thread not synchronized	Use _coordinator.is_shutting_down flag to drain remaining messages before closing
SQLite lock contention	Checkpoint write blocks WebSocket receive thread	Checkpoint written on every message (too frequent)	Write checkpoint on a timer (every N seconds) or only on shutdown; never on every message

---

Closing

The gap between a reliable data pipeline and a fragile one is not a different architecture—it is five lines of signal handling, a SQLite table, and a gap-fill query. The pattern is simple enough to implement in an afternoon, and the failure modes it prevents—silent data loss, duplicate processing, missed signals—are exactly the kind that erode confidence in a trading system over months before anyone notices.

The core insight is this: your pipeline's state is not in memory. It is on disk. Every 30 seconds, write the checkpoint. On startup, read it. On shutdown, flush it. The rest is just WebSocket plumbing.

---

Next Steps

If you're building a real-time data pipeline:

1. Clone the checkpoint manager and signal handler modules from this article
2. Integrate them into your existing WebSocket client
3. Set the TICKDB_API_KEY environment variable, then run the startup sequence

If you need 10+ years of historical OHLCV data for backtesting your strategy:
Reach out to enterprise@tickdb.ai for institutional data plans covering US equities, HK stocks, and crypto.

If you're integrating TickDB into an AI-assisted workflow:
Search for and install the tickdb-market-data SKILL in your AI tool's marketplace for code generation assistance.

---

This article does not constitute investment advice. Markets involve risk; past performance does not guarantee future results.