Payment Channels and Layer-2 Protocols

Base-layer blockchains settle every transaction on every node. This provides the strongest security guarantees but limits throughput to single-digit or low-double-digit transactions per second. Payment channels solve this by moving the vast majority of transactions off-chain while inheriting the base layer’s security through cryptographic commitments.

The core insight: two parties who transact frequently don’t need to broadcast every payment to the entire network. They can maintain a private ledger between themselves, secured by pre-signed transactions that could be broadcast to the blockchain if needed, but never actually are during normal operation.

The diagram above shows the state machine for a bidirectional payment channel. Each state transition is backed by a signed Bitcoin transaction that could be broadcast to close the channel at any time.

Payment Channel Lifecycle

Opening a Channel

Opening a channel requires one on-chain transaction that locks funds into a 2-of-2 multisig address:

from dataclasses import dataclass
from decimal import Decimal

@dataclass
class ChannelState:
    """
    Represents the current state of a bidirectional payment channel.
    
    Both parties maintain a copy of the current state. Each state
    update is a signed commitment transaction that redistributes
    the channel's total capacity between the two parties.
    """
    channel_id: bytes
    party_a: str             # Public key of party A
    party_b: str             # Public key of party B
    capacity_sat: int        # Total channel capacity (locked on-chain)
    balance_a_sat: int       # Party A's current balance
    balance_b_sat: int       # Party B's current balance
    state_number: int        # Monotonically increasing state counter
    commitment_tx_a: bytes   # Commitment transaction signed by B (held by A)
    commitment_tx_b: bytes   # Commitment transaction signed by A (held by B)
    
    @property
    def is_balanced(self) -> bool:
        return self.balance_a_sat + self.balance_b_sat == self.capacity_sat

class PaymentChannel:
    """
    Bidirectional payment channel with revocation-based fraud protection.
    
    This implements the LN-penalty mechanism used in Lightning Network:
    - Each state update creates a new commitment transaction
    - The previous state is "revoked" by exchanging revocation keys
    - If a party broadcasts a revoked state, the counterparty can
      claim ALL funds in the channel (the "penalty")
    """
    
    def __init__(self, state: ChannelState):
        self._state = state
        self._revoked_states: list[ChannelState] = []
    
    def create_update(
        self, sender: str, amount_sat: int
    ) -> ChannelState:
        """
        Create a new channel state transferring amount from sender.
        
        This is the core operation: it produces a new commitment
        transaction pair without touching the blockchain.
        
        In Lightning, this takes ~1ms (just signing) vs ~10 minutes
        for an on-chain Bitcoin transaction.
        """
        if sender == self._state.party_a:
            if self._state.balance_a_sat < amount_sat:
                raise InsufficientChannelBalance(
                    f"A has {self._state.balance_a_sat}, needs {amount_sat}"
                )
            new_balance_a = self._state.balance_a_sat - amount_sat
            new_balance_b = self._state.balance_b_sat + amount_sat
        elif sender == self._state.party_b:
            if self._state.balance_b_sat < amount_sat:
                raise InsufficientChannelBalance(
                    f"B has {self._state.balance_b_sat}, needs {amount_sat}"
                )
            new_balance_a = self._state.balance_a_sat + amount_sat
            new_balance_b = self._state.balance_b_sat - amount_sat
        else:
            raise ValueError(f"Unknown sender: {sender}")
        
        # Archive current state as revoked
        self._revoked_states.append(self._state)
        
        new_state = ChannelState(
            channel_id=self._state.channel_id,
            party_a=self._state.party_a,
            party_b=self._state.party_b,
            capacity_sat=self._state.capacity_sat,
            balance_a_sat=new_balance_a,
            balance_b_sat=new_balance_b,
            state_number=self._state.state_number + 1,
            commitment_tx_a=self._build_commitment(new_balance_a, new_balance_b, "a"),
            commitment_tx_b=self._build_commitment(new_balance_a, new_balance_b, "b"),
        )
        
        self._state = new_state
        return new_state
    
    def _build_commitment(
        self, balance_a: int, balance_b: int, holder: str
    ) -> bytes:
        """
        Build a commitment transaction.
        
        Each party holds a slightly different version:
        - A's version: A's output has a timelock (can't spend for N blocks)
        - B's version: B's output has a timelock
        
        The timelock gives the counterparty time to submit a fraud
        proof if a revoked state is broadcast.
        """
        # Simplified — in production this creates a real Bitcoin transaction
        return (
            f"COMMIT state={self._state.state_number + 1} "
            f"A={balance_a} B={balance_b} holder={holder}"
        ).encode()
    
    def close_cooperative(self) -> bytes:
        """
        Cooperative close: both parties agree on final balances.
        
        Produces a single on-chain transaction with no timelocks.
        This is the happy path — used in ~95% of channel closures.
        """
        return (
            f"CLOSE channel={self._state.channel_id.hex()} "
            f"A={self._state.balance_a_sat} "
            f"B={self._state.balance_b_sat}"
        ).encode()
    
    def close_unilateral(self, closer: str) -> bytes:
        """
        Unilateral close: one party broadcasts their commitment tx.
        
        The closing party's output is timelocked (typically 144 blocks
        = ~1 day). This gives the counterparty time to check for fraud.
        """
        if closer == self._state.party_a:
            return self._state.commitment_tx_a
        return self._state.commitment_tx_b
    
    def detect_fraud(self, broadcast_state_number: int) -> bool:
        """
        Check if a broadcast commitment uses a revoked state.
        
        If fraud is detected, the honest party can claim ALL channel
        funds using the revocation key for that state.
        """
        return broadcast_state_number < self._state.state_number

class InsufficientChannelBalance(Exception):
    pass

HTLC: Trustless Multi-Hop Payments

A payment channel only works between two direct counterparties. The Lightning Network extends this to a network of channels using Hash Time-Locked Contracts (HTLCs):

import hashlib
import os
import time
from dataclasses import dataclass

@dataclass
class HTLC:
    """
    Hash Time-Locked Contract: conditional payment.
    
    "I will pay you X if you reveal the preimage of hash H
     within T seconds. Otherwise, I get my money back."
    
    This enables trustless multi-hop payments:
    1. Recipient generates a random secret R and shares hash(R)
    2. Each hop in the route creates an HTLC locked to hash(R)
    3. Recipient reveals R to claim the final hop's payment
    4. Each intermediate node uses the revealed R to claim their
       incoming HTLC
    
    The time locks decrease at each hop, ensuring that intermediate
    nodes always have time to claim their incoming payment after
    forwarding the outgoing one.
    """
    payment_hash: bytes     # SHA-256 hash of the preimage
    amount_sat: int         # Amount locked
    expiry_time: int        # Unix timestamp — refund after this time
    sender: str
    receiver: str
    
    def can_claim(self, preimage: bytes) -> bool:
        """Check if the preimage unlocks this HTLC."""
        return hashlib.sha256(preimage).digest() == self.payment_hash
    
    def is_expired(self) -> bool:
        return time.time() > self.expiry_time


class LightningRouter:
    """
    Find a route through the Lightning Network for a payment.
    
    The routing problem in Lightning is a variant of shortest-path
    with constraints:
    - Each channel has a capacity (max payment size)
    - Each channel has a fee schedule (base_fee + fee_rate * amount)
    - Channel balances are private (sender doesn't know exact balances)
    - Time locks accumulate along the route
    """
    
    def __init__(self):
        # Graph: node -> [(neighbor, channel_info)]
        self._graph: dict[str, list[dict]] = {}
    
    def add_channel(
        self, node_a: str, node_b: str, capacity_sat: int,
        base_fee_msat: int, fee_rate_ppm: int, timelock_delta: int
    ):
        """
        Add a channel to the routing graph.
        
        fee_rate_ppm: fee rate in parts per million
        timelock_delta: CLTV delta (blocks) required by this hop
        """
        channel = {
            "peer": node_b,
            "capacity": capacity_sat,
            "base_fee_msat": base_fee_msat,
            "fee_rate_ppm": fee_rate_ppm,
            "timelock_delta": timelock_delta,
        }
        
        if node_a not in self._graph:
            self._graph[node_a] = []
        self._graph[node_a].append(channel)
        
        # Channels are bidirectional
        reverse = dict(channel)
        reverse["peer"] = node_a
        if node_b not in self._graph:
            self._graph[node_b] = []
        self._graph[node_b].append(reverse)
    
    def find_route(
        self, source: str, destination: str, amount_sat: int
    ) -> list[dict]:
        """
        Find the cheapest route from source to destination.
        
        Uses a modified Dijkstra's algorithm that:
        1. Works backwards from destination to source
        2. Accumulates fees at each hop (fee depends on forwarded amount)
        3. Filters channels with insufficient capacity
        
        Working backwards is critical: each hop's fee depends on the
        amount being forwarded, which depends on subsequent hops' fees.
        """
        import heapq
        
        # Dijkstra from destination to source (backward search)
        dist = {destination: amount_sat}
        prev = {}
        visited = set()
        heap = [(amount_sat, destination)]
        
        while heap:
            cost, node = heapq.heappop(heap)
            
            if node in visited:
                continue
            visited.add(node)
            
            if node == source:
                break
            
            # Explore neighbors (in reverse direction)
            for neighbor_node in self._graph:
                for channel in self._graph[neighbor_node]:
                    if channel["peer"] != node:
                        continue
                    if neighbor_node in visited:
                        continue
                    
                    # Calculate fee for forwarding 'cost' satoshis
                    fee = (
                        channel["base_fee_msat"] / 1000 +
                        cost * channel["fee_rate_ppm"] / 1_000_000
                    )
                    total_needed = int(cost + fee)
                    
                    # Check capacity
                    if total_needed > channel["capacity"]:
                        continue
                    
                    if (neighbor_node not in dist or 
                        total_needed < dist[neighbor_node]):
                        dist[neighbor_node] = total_needed
                        prev[neighbor_node] = (node, channel)
                        heapq.heappush(heap, (total_needed, neighbor_node))
        
        if source not in prev and source != destination:
            return []  # No route found
        
        # Reconstruct path
        route = []
        current = source
        while current != destination:
            next_node, channel = prev[current]
            route.append({
                "from": current,
                "to": next_node,
                "amount": dist[current],
                "fee_msat": int(
                    channel["base_fee_msat"] + 
                    dist[current] * channel["fee_rate_ppm"] / 1_000_000
                ),
                "timelock_delta": channel["timelock_delta"],
            })
            current = next_node
        
        return route


def execute_multi_hop_payment(
    route: list[dict], payment_preimage: bytes
) -> list[HTLC]:
    """
    Construct the HTLC chain for a multi-hop Lightning payment.
    
    Example route: Alice → Bob → Carol → Dave
    
    1. Dave gives Alice hash(preimage)
    2. Alice creates HTLC to Bob: "Pay Bob X+fees if he knows preimage, 
       expires in 40 blocks"
    3. Bob creates HTLC to Carol: "Pay Carol X+carol_fee if she knows
       preimage, expires in 30 blocks"
    4. Carol creates HTLC to Dave: "Pay Dave X if he knows preimage,
       expires in 20 blocks"
    5. Dave reveals preimage to Carol (claims her HTLC)
    6. Carol uses preimage to claim Bob's HTLC
    7. Bob uses preimage to claim Alice's HTLC
    
    The decreasing timelocks ensure each forwarder has time to claim
    their incoming HTLC after learning the preimage from forwarding.
    """
    payment_hash = hashlib.sha256(payment_preimage).digest()
    base_timelock = int(time.time()) + 3600  # 1 hour from now
    
    htlcs = []
    cumulative_timelock = 0
    
    for i, hop in enumerate(route):
        cumulative_timelock += hop["timelock_delta"] * 600  # ~10 min/block
        
        htlc = HTLC(
            payment_hash=payment_hash,
            amount_sat=hop["amount"],
            expiry_time=base_timelock + (len(route) - i) * hop["timelock_delta"] * 600,
            sender=hop["from"],
            receiver=hop["to"],
        )
        htlcs.append(htlc)
    
    return htlcs

Rollups: Scaling Payments with Validity Proofs

Payment channels require capital lockup and online participation. Rollups take a different approach: batch hundreds of payments into a single on-chain transaction, using either fraud proofs (Optimistic rollups) or validity proofs (ZK rollups) to ensure correctness.

@dataclass
class RollupPayment:
    """A payment within a rollup batch."""
    sender_index: int       # Index in the rollup's account tree
    receiver_index: int     # Index in the rollup's account tree
    amount: int
    nonce: int
    signature: bytes

class PaymentRollup:
    """
    A simplified ZK-Rollup for payments.
    
    Architecture:
    1. Users submit payments to the rollup operator
    2. Operator batches payments (e.g., 1000 per batch)
    3. Operator computes the new state root after applying all payments
    4. Operator generates a ZK proof that the state transition is valid
    5. Operator posts (new_root, proof, compressed_data) on-chain
    
    The on-chain contract verifies the proof and updates the state root.
    If the proof is valid, the batch is finalized in one L1 transaction.
    
    Throughput: 1000-10000 TPS (vs 7 for Bitcoin, 30 for Ethereum)
    Cost: amortized over the batch (e.g., 1000 payments share one L1 tx fee)
    Finality: L1 finality (the proof guarantees correctness)
    """
    
    def __init__(self, num_accounts: int = 2**20):
        self._balances: list[int] = [0] * num_accounts
        self._nonces: list[int] = [0] * num_accounts
        self._batch: list[RollupPayment] = []
        self._batch_limit = 1000
    
    def submit_payment(self, payment: RollupPayment) -> bool:
        """Add a payment to the current batch."""
        # Validate off-chain (fail fast)
        if payment.sender_index >= len(self._balances):
            return False
        if payment.nonce != self._nonces[payment.sender_index]:
            return False
        if self._balances[payment.sender_index] < payment.amount:
            return False
        
        self._batch.append(payment)
        
        if len(self._batch) >= self._batch_limit:
            self._process_batch()
        
        return True
    
    def _process_batch(self):
        """
        Process the batch: apply all payments and generate proof.
        
        In production, the proof generation uses a ZK circuit that
        encodes the payment validation rules. The circuit verifies:
        - Each payment's signature is valid
        - Each sender has sufficient balance
        - Nonces are correct
        - The new state root is correctly computed
        
        Proof generation takes 1-10 minutes for a batch of 1000.
        Verification takes ~1ms on the L1 contract.
        """
        pre_state_root = self._compute_state_root()
        
        for payment in self._batch:
            self._balances[payment.sender_index] -= payment.amount
            self._balances[payment.receiver_index] += payment.amount
            self._nonces[payment.sender_index] += 1
        
        post_state_root = self._compute_state_root()
        
        # Compress payment data for on-chain storage
        # Each payment compresses to ~12 bytes:
        # sender_index (3B) + receiver_index (3B) + amount (4B) + fee (2B)
        compressed = self._compress_batch()
        
        batch_submission = {
            "pre_state_root": pre_state_root,
            "post_state_root": post_state_root,
            "num_payments": len(self._batch),
            "compressed_data_bytes": len(compressed),
            "proof": self._generate_validity_proof(),
        }
        
        self._batch.clear()
        return batch_submission
    
    def _compute_state_root(self) -> bytes:
        """Compute Merkle root of the account state."""
        # Simplified — production uses a sparse Merkle tree
        import hashlib
        data = b''.join(
            (bal.to_bytes(8, 'big') + nonce.to_bytes(4, 'big'))
            for bal, nonce in zip(self._balances, self._nonces)
        )
        return hashlib.sha256(data).digest()
    
    def _compress_batch(self) -> bytes:
        """Compress batch data for on-chain calldata."""
        result = bytearray()
        for p in self._batch:
            result.extend(p.sender_index.to_bytes(3, 'big'))
            result.extend(p.receiver_index.to_bytes(3, 'big'))
            result.extend(p.amount.to_bytes(4, 'big'))
        return bytes(result)
    
    def _generate_validity_proof(self) -> bytes:
        """Generate ZK proof of batch validity."""
        # In production: uses Groth16 or PLONK proving system
        return b"PROOF_PLACEHOLDER"

Throughput Comparison

Lightning achieves the highest theoretical throughput because channel updates are just signature exchanges between two parties — they don’t touch any shared infrastructure. ZK-Rollups offer a middle ground: high throughput without requiring capital lockup or online participation.

For a payment system architect choosing between these approaches: Lightning is ideal for high-frequency, low-value payments between parties who transact regularly. ZK-Rollups are better for systems with many diverse participants who don’t have pre-existing relationships. Both inherit the security of their base layer — a property that no traditional payment rail can claim without trusting intermediaries.