Reactive Live Data, Distributed State & CDC

Spedo couples in-process reactive state with one central RESP store. It provides

Live Variables, optional Change Data Capture (CDC), and a Write-Behind helper

for external SQL databases. These are preview features, not a replacement for

replication, durable messaging, or cross-node high availability.

1. Distributed Live Variables (`SpedoLiveVar`)

A SpedoLiveVar behaves like a regular Python variable, but its value can be

coordinated between application instances that connect to the same Spedo node.

It does not make the underlying node highly available.

Reading .value returns the local in-process copy. On an invalidation, the client reloads the key from Spedo before notifying callbacks. The first load and each invalidated reload are network operations; an unchanged local read is not.

The data path is push invalidation + pull refresh:

1. assigning .value updates the writer's local copy and sends SET;

2. Spedo publishes the invalidated key name through SPEDO.WATCH;

3. other bound clients issue one GET, replace their local copy, then notify

callbacks and async waiters;

4. subsequent .value reads stay inside the Python process.

default is a local fallback when the server key does not exist; binding does

not create that key. Assign .value explicitly to publish an initial value.

For dictionaries and lists, assign a new value instead of mutating

bound.value[...] in place, because only assignment invokes the setter and

sends SET.

Usage Example

from spedo_client import SpedoClient

client = SpedoClient(host="127.0.0.1", port=6380)

# Bind a reactive variable to a key. `bind_var` is the current SDK API.
feature_flag = client.bind_var("flags:new_checkout_flow", default=False)

# Local memory read when the cached value is current
if feature_flag.value:
    print("New checkout enabled!")

# When any pod modifies the value:
feature_flag.value = True
# Other connected pods receive an invalidation and reload the value.

Real-world example: a live payment kill switch

from spedo_client import SpedoClient

api = SpedoClient(host="spedo", port=6380)
control = SpedoClient(host="spedo", port=6380)

payment_policy = api.bind_var(
    "runtime:payments",
    default={"enabled": True, "max_amount": 2500},
)
control_policy = control.bind_var(
    "runtime:payments",
    default={"enabled": True, "max_amount": 2500},
)

@payment_policy.on_change
def policy_changed(policy):
    print("Payment policy reloaded:", policy)

# One control-plane SET. Connected API processes invalidate and reload once.
control_policy.value = {
    "enabled": False,
    "max_amount": 0,
    "reason": "provider incident",
}

def authorize_payment(amount):
    policy = payment_policy.value  # local in-process read on the hot path
    if not policy["enabled"]:
        return {"status": "temporarily_unavailable"}
    verdict = "accepted" if amount <= policy["max_amount"] else "review"
    return {"status": verdict}

The v0.41 feature harness measured 1,579,176 local .value reads/s versus

1,497 Redis TCP GET calls/s in its Python loops. This is an application-path

comparison: the gain comes from removing TCP from repeated reads, while each

change still pays one server write and one reload per connected binding.

Async Waiting for State Changes

import asyncio

async def monitor_task():
    # Wait until status becomes "ready"
    await task_status.wait_until("ready", timeout=10.0)
    print("Task is now ready!")

    # Or wait for any change
    new_val = await task_status.wait_change(timeout=5.0)
    print(f"Status changed to: {new_val}")

Python class instances (`serializer="pickle"`)

json is the default serializer. Use pickle when the distributed value is a

real Python instance rather than a JSON-compatible value:

from dataclasses import dataclass

@dataclass
class Account:
    id: int
    roles: list[str]

account = client.bind_var(
    "account:42",
    default=Account(42, ["read"]),
    serializer="pickle",
    trusted=True,
)

trusted=True is required to restore application classes. Enable it only for

keys written exclusively by trusted code: a hostile pickle can execute code on

load. Built-in values (dict, list, strings, numbers, bytes…) can use

serializer="pickle" without trusted=True.

bind_live is not a current SDK method; older website examples used that name.

Use bind_var in new code.

2. Zero-Boilerplate Auto-Sync Globals (`from spedo import auto_sync`)

auto_sync is a convenience layer for small, low-contention module globals.

It wraps functions already defined in the module and performs a **pull before

each wrapped call, then a push after a detected change** to one Spedo node. It

is not a subscription protocol, a distributed transaction, or a cross-instance

consistency guarantee.

How it works

At auto_sync(globals()) time, it inspects each module function's compiled

bytecode (func.__code__.co_names) and wraps the functions it finds. A

function defined later is not wrapped automatically. co_names is a

bytecode name list, not a semantic declaration of writable globals, so keep

wrapped functions narrow and avoid assuming every discovered name is a

business-state variable.

Before a wrapped function runs, it issues GET global:<name> for each

discovered name and replaces the local global only when a remote value

exists.

After the function returns, it compares the pre-call and post-call values;

changed values are written with SET global:<name>.

Change detection keeps the pre-call object reference. Reassign a new

dict/list value when it changes; an in-place mutation can compare equal

to that same reference and therefore may not be pushed.

There is no background watcher, pub/sub invalidation, lock, compare-and-set,

or transaction. Other instances see an update only when one of their own

wrapped calls starts and pulls from Spedo.

Concurrent read-modify-write operations are therefore non-atomic. If two

instances pull the same value and both write a replacement, the last SET

accepted by the server wins and an increment can be lost. Use an explicit

atomic server command or a workload-specific coordination design for counters,

money, reservations, or any correctness-sensitive shared state.

dict and list values are JSON-encoded; other values are converted to text.

Do not use auto_sync for arbitrary Python objects or values that need exact

type preservation.

Usage Example:

from spedo import auto_sync

# 1. Declare standard Python variables (NO wrappers, NO special objects)
payment_total = 0
active_orders = 0
promo_banner  = "WELCOME_2026"

# 2. Write regular Python functions.  They must exist before auto_sync() runs.
def process_payment(amount: int):
    global payment_total, active_orders
    payment_total = int(payment_total) + amount
    active_orders = int(active_orders) + 1

def update_promo(new_banner: str):
    global promo_banner
    promo_banner = new_banner

# 3. One call wraps the functions currently defined in this module.
# Each wrapped call pulls first and pushes detected changes afterward.
auto_sync(globals())

# This is a non-atomic read-modify-write operation: do not use it for a counter
# whose updates must never be lost under concurrent writers.
process_payment(50)

3. Distributed Live Dictionaries (`SpedoLiveDict`)

Synchronize complex structured dictionaries with fine-grained field updates across pods:

cluster_config = client.bind_dict("config:cluster_nodes", default={"master": "node-1"})

# Read keys locally in 0 ms
print(cluster_config["master"])

# Update dictionary keys
cluster_config["worker_count"] = 8
cluster_config.update({"region": "eu-west-1", "active": True})

# Mutate nested dictionary
with cluster_config.mutate() as d:
    d["last_heartbeat"] = 1787255000

3. Native Change Data Capture (CDC)

CDC (Change Data Capture) is an ordered in-memory event stream of mutations

(SET, DEL, etc.) with timestamps and sequence IDs. It is **disabled by

default because recording each write has a cost. Enable cdc in Admin →

Observability & Debug**, or start the server with SPEDO_CDC_ENABLED=true.

The log is bounded and non-persistent: it is not a backup and consumers must

keep their checkpoint / handle missed history. CDC is bounded and in-memory:

it is not a durable event log or recovery mechanism.

Querying the CDC Log

# Get the latest sequence ID
latest_seq = client.cdc_latest()

# Read the last 50 mutation events starting from a sequence ID
events, new_latest = client.cdc_read(last_id=0, count=50)

for event in events:
    print(
        f"[SEQ #{event['id']}] "
        f"OP={event['op']} | "
        f"KEY={event['key']} | "
        f"VAL={event['value']} | "
        f"TS={event['timestamp_ms']}"
    )

4. Asynchronous Write-Behind to SQL Databases

With Write-Behind synchronization, application writes are sent to Spedo while a

background worker consumes the CDC stream in micro-batches and persists records

to PostgreSQL, SQLite, MySQL, or ClickHouse. It requires CDC to be enabled.

Write-behind enqueue throughput and the server ACK throughput are different

measurements; use an explicit flush/error policy when durability matters.

import sqlite3
import json

# Setup external SQL database
db_conn = sqlite3.connect("app.db", check_same_thread=False)

def sync_mutations_to_sql(batch_events):
    """Callback invoked by the background worker for each batch of mutations."""
    cursor = db_conn.cursor()
    for ev in batch_events:
        key = ev["key"]
        if key.startswith("user:"):
            user_id = key.split(":")[-1]
            if ev["op"] == "SET" and ev["value"]:
                val_str = ev["value"].decode() if isinstance(ev["value"], bytes) else str(ev["value"])
                cursor.execute(
                    "INSERT INTO users (id, payload, updated_at) VALUES (?, ?, ?) "
                    "ON CONFLICT(id) DO UPDATE SET payload=excluded.payload, updated_at=excluded.updated_at",
                    (user_id, val_str, ev["timestamp_ms"])
                )
            elif ev["op"] == "DEL":
                cursor.execute("DELETE FROM users WHERE id = ?", (user_id,))
    db_conn.commit()

# Start the background sync worker
worker = client.db_sync_worker(
    on_mutation=sync_mutations_to_sql,
    poll_interval=0.05, # Check every 50ms
    batch_size=100
).start()

# Application writes only to Spedo (ultra-fast, zero SQL bottleneck)
client.set("user:101", json.dumps({"name": "Alice", "status": "active"}))
client.set("user:102", json.dumps({"name": "Bob", "status": "pro"}))

# The worker automatically writes records into the SQL database in the background!

# Graceful shutdown when application stops
worker.stop()